Utilizando o FastAPI – parte 3

utilizando-fastapi-3_abertura

Dando sequência ao desenvolvimento da API, a primeira parte introduziu o FastAPI e implementou os métodos GET e DELETE, a segunda parte acrescentou o pydantic e adicionou os métodos POST e também o PUT. Porém a API ainda trabalha com um arquivo JSON de “banco de dados”.

Esta parte tem um pouco de FastAPI mas consiste basicamente na inclusão do SQLAlchemy dentro do projeto para permitir acesso a bancos de dados SQL via uma ORM.

SQLAlchemy

O SQLAlchemy é uma biblioteca em Python que implementa acesso a bancos de dados SQL e também um mapeamento de objeto relacional (ou ORM de object relational mapping). Ou seja, implementa tanto as abstrações necessárias para a conexão ao banco de dados em si como também a manipulação de seu conteúdo como uma estrutura de dados nativa da linguagem.

Ah sim, a ideia aqui não foi escrever um tutorial sobre como usá-lo, inclusive fui bem superficial em algumas partes mas, procurando por algo do tipo, recomendo a leitura do tutorial básico de SQLAlchemy da Letícia Portela.

Instalando o SQLAlchemy

Começando com a instalação do SQLAlchemy, que é feita através do pip.

$ pip install sqlalchemy
Collecting sqlalchemy
  Downloading SQLAlchemy-1.3.19- ... cp38-cp38-(...).whl (1.3 MB)
     |████████████████████████████████| 1.3 MB 234 kB/s 
Installing collected packages: sqlalchemy
Successfully installed sqlalchemy-1.3.19

Depois é adicioná-lo, também, na lista de dependências do projeto dentro do arquivo “requirements.txt”.

Adicionando um banco de dados

Como não conhecia muito bem o SQLAlchemy acabei seguindo o exemplo usado que é usado na documentação do FastAPI e também separei a implementação do banco de dados em três arquivos distintos:

  • database.py
  • models.py e
  • crud.py

O motivo está justamente na divisão das responsabilidades, assim, para modificar o banco de dados mexe-se apenas no primeiro, para adicionar ou remover campos altera-se apenas o segundo e talvez o terceiro (dependendo da mudança) e por aí vai… 🙂

Conexão com o banco de dados

A conexão com o banco de dados é definida dentro de “databases.py” e como preferi deixar as coisas simples, optei pelo SQLite3 mesmo.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine(
    "sqlite:///db.sqlite3",
    connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(
    autocommit=False, autoflush=False, bind=engine
)

Base = declarative_base()

A primeira coisa a fazer é definir o engine a utilizar, ou seja, com qual SGBD (dialeto) o SQLAlchemy trabalhará e os parâmetros de conexão para ele — na documentação do SQLAlchemy é possível consultar quais os dialetos suportados por ele, tanto de forma nativa como por bibliotecas externas.

Aqui apenas defino como será feita a conexão, a sessão em si será realizada em outro ponto do código.

Criando o modelo

O modelo de dados está em “models.py” (dah!) e é literalmente uma transcrição daquele já definido para o pydantic dentro do arquivo “schemas.py”.

from sqlalchemy import Column, Integer, String
from .database import Base

class Student(Base):
    __tablename__ = "students"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    address = Column(String)
    neighbour = Column(String)
    city = Column(String)
    state = Column(String)
    postal_code = Column(String)

E, sendo repetitivo, para deixar esta implementação simples preferi manter um modelo de tabela única.

Manipulação dos registros

Os registros são manipulados apenas dentro de “crud.py”, que cuida das operações básicas de inserção, atualização, atualização e remoção de registos — o “tal” do CRUD.

from typing import Generator

from sqlalchemy.orm import Session

from .datatypes import UpdateStudentValuesType
from .models import Student
from .schemas import CreateStudentSchema, StudentSchema

students = Student
...

Além dos arquivos já definidos acima ele também importa elementos de “datatypes.py” (que contém anotações de tipo customizadas) e de “schemas.py” (o modelo do pydantic).

Todas as funções recebem a sessão com o banco de dados já estabelecida como um dos seus argumentos e, sim, a conexão com o banco de dados não é feita aqui! 😀

Criando um novo registro

A criação de um novo registro é simples, os dados são recebidos dentro de “student” que graças às anotações de tipo sabemos ser um modelo do pydantic que já chega validado e do qual conhecemos o conteúdo.

def create_student(
    db: Session, student: CreateStudentSchema
):
    new_student = Student(**student.dict())
    db.add(new_student)
    db.commit()
    db.refresh(new_student)
    return new_student

Os dados do novo estudante são convertidos para um dicionário, método dict(), usado para criar um novo registro na tabela de estudantes e então adicionado ao banco de dados — os métodos add() e commit() cuidam desta parte.

O método refresh() é usado para atualizar o conteúdo do registro (recuperar o valor do id, por exemplo) antes de retorná-lo.

Recuperando registros

Há duas funções para recuperar os registros, uma pega todos aqueles que estão disponíveis.

def retrieve_all_students(db: Session) -> Generator:
    return db.query(students).all()

Ela consulta a tabela de estudantes e usa o método all() para recuperar todos os registros e enviá-los dentro de um gerador. Só lembrando que a paginação para a consulta não foi implementada então todos os registros são enviados de uma vez só.

Já a outra função retorna apenas um registro e de id específico, informado em “student_id”.

def retrieve_student(db: Session, student_id: int):
    return db.query(students).filter(
        students.id == student_id
    ).first()

Ela faz o mesmo tipo de chamada que a função anterior, também usa o método all(), porém aplica o método filter() para selecionar aprnas os registros corretos² e retornar a primeira ocorrência deste via first().

Atualizando um registro

A atualização dos registros recebe além da sessão e do id, um dicionário contendo os valores que precisam ser alterados no registro.

def update_student(
    db: Session,
    student_id: int,
    values: UpdateStudentValuesType
):
    if student := retrieve_student(db, student_id):
        db.query(students).filter(
            students.id == student_id
        ).update(values)
        db.commit()
        db.refresh(student)

        return student

Ela funciona em duas etapas, a primeira verifica/recupera os dados do estudante enquanto que a segunda parte altera os valores do registro efetuando uma segunda consulta que usa o métodos update() para substituir os valores. O método commit() persiste as alterações e o refresh() atualiza os dados em “student”.

Apagando um registro

Por fim, a rotina que apaga registros e ela também usa a função retrive_student() para validar a existência do estudante.

def remove_student(db: Session, student_id: int) -> bool:
    if student := retrieve_student(db, student_id):
        db.delete(student)
        db.commit()
        return True

    return False

Existindo ela apaga o registro com os métodos delete() e commit() e retorna True para indicar o sucesso da operação. Caso o registro não exista ele retornará False.

(²) Aqueles cujo o valor do id é igual a “student_id” e ao qual esperamos que seja o único.

Modificando a API

Agora que as rotinas de acesso ao banco de dados estão disponíveis é hora de remover o acesso ao banco de dados de mentira, começando pelas funções get_max() e retrieve_students() e também o suporte para arquivos JSON como um todo lá no “main.py” visto que não são mais necessários.

Depois é adicionar o SQLAlchemy e também o resto da estrutura definida mais acima.

from fastapi import (
    Depends,
    FastAPI,
    HTTPException,
    Response,
    status,
)
from sqlalchemy.orm import Session
from .crud import (
    create_student,
    remove_student,
    retrieve_all_students,
    retrieve_student,
    update_student,
)
from .database import Base, SessionLocal, engine
from .datatypes import StudentType
from .schemas import (
    CreateStudentSchema,
    StudentSchema,
    UpdateStudentSchema,
)
...

E, então…

Base.metadata.create_all(bind=engine)


def get_db() -> Generator:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

O primeiro comando cuida de criar no banco de dados as tabelas definidas como no modelo (no caso do SQLite3 ele criará inclusive o arquivo “db.sqlite3”). Já a função get_db() é encarregada de estabelecer a conexão com o banco de dados que é passada (via injeção de dependência) para cada função que precisa conectar ao banco.

Cmo grande parte da lógica para manipular o banco de dados estão em outras arquivos as funções agora em “main.py” cuidam basicamente da implementação do método HTTP. Aliás, aproveitei que mexeria nas funções para corrigir os códigos de retorno do HTTP e deixá-las mais de acordo com as recomendações da RFC7231.

O argumentos “response_model” no decorador foram removidos por estare conflitando com a validação do pydantic.

Os métodos GET

A função que implementa o método de recuperação de todos os estudantes cadastrados ficou assim:

@app.get(
    "/students/",
    status_code=status.HTTP_200_OK,
)
def get_all_students(
    db: Session = Depends(get_db)
) -> Generator:
    if result := retrieve_all_students(db):
        return result

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Não existem estudantes cadastrados.",
    )

Basicamente ela recebe a conexão com o banco de dados e apenas chama a função retrieve_all_students() de “crud.py”.

Já a que implementa o método de recuperação de apenas um estudante ficou desta forma.

@app.get(
    "/students/{student_id}/",
    status_code=status.HTTP_200_OK,
)
def get_student(
    student_id: int,
    db: Session = Depends(get_db)
) -> StudentType:
    if result := retrieve_student(db, student_id):
        return result

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"Estudante de 'id={student_id}' não encontrado.",
    )

E para ambos os métodos, na falta de conteúdo, ele gerará uma exceção de HTTP 404 (não encontrado).

O método DELETE

O método que remove estudantes ficou desta maneira.

@app.delete(
    "/students/{student_id}/",
    status_code=status.HTTP_204_NO_CONTENT
)
def delete_student(
    student_id: int,
    db: Session = Depends(get_db)
) -> None:
    if not remove_student(db, student_id):
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Estudante de 'id={student_id}' não encontrado.",
        )

Recebe o id do estudante a ser removido e chama a função remove_student(), retornando False será produzida uma exceção de HTTP 404 (não encontrado) mas em caso de True não faz nada e deixa para o FastAPI sozinho cuidar para retornar HTTP 201 (sem conteúdo).

O método POST

O método para inclusão de estudantes ficou assim.

@app.post(
    "/students/",
    status_code=status.HTTP_201_CREATED,
)
def post_student(
    student: CreateStudentSchema,
    db: Session = Depends(get_db),
) -> StudentType:
    if result := create_student(db, student):
        return result

    raise HTTPException(
        status_code=status.HTTP_400_BAD_REQUEST
    )

Recebe os dados já validados pelo pydantic e envia para a função create_student() e devolve o resultado (basicamente o que foi enviado acrescido do campo de id) em HTTP 201 (criado), Mas se algo de errado ocorrer ele gera uma exceção de HTTP 400 (requisição ruim — numa tradução literal).

O método PUT

O método para alteração de estudantes parece uma versão sofisticada do método de inserção.

@app.put(
    "/students/{student_id}",
    status_code=status.HTTP_201_CREATED,
)
def put_student(
    student_id: int,
    student: UpdateStudentSchema,
    db: Session = Depends(get_db),
) -> StudentType:
    if result := update_student(
        db, student_id, {
            key: value for key, value in student if value
        }
    ):
        return result

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"Estudante de 'id={student_id}' não encontrado.",
    )

Também recebe os dados já validados pelo pydantic mas antes de chamar a função update_student(), remove eventuais campos com valores vazios dela, daí retorna HTTP 201 (criado) junto com o registro atualizado ou em caso o estudante não existente gera exceção de HTTP 404 (não encontrado).

Finalizando esta parte…

Aproveitei para criar um diretório, o “./api”, e colocar toda a aplicação lá dentro em virtude da quantidade de arquivos que tem agora. E em “./examples/students.sql” está o arquivo SQL contendo os dados do arquivo JSON e para recriar a base de dados, use:

$ sqlite3 db.sqlite3 < examples/students.sql

Na próxima parte serão implementadas as rotinas de teste unitário (não estava no escopo original mas achei interessante acrescentá-los).

Os arquivos desta parte já estão disponíveis no seu respectivo repositório no GitHub na tag “parte-3”.

Até!

Um comentário sobre “Utilizando o FastAPI – parte 3

  1. Pingback: Utilizando o FastAPI – parte 4 | giovannireisnunes

Os comentários estão desativados.