Tratamento de sessão eficiente usando o padrão de repositório no FastAPI

Escrevi um artigo há cerca de cinco meses sobre como adaptar o padrão de repositório com FastAPI e recebi muitas leituras (obrigado). Eu vim escrever sobre uma maneira eficiente de lidar com o tratamento de sessão, ainda usando o padrão de repositório.

Antes de ir direto ao assunto, notei que, na produção, recebo um dos seguintes erros sempre que minhas APIs estão tentando fazer uma transação que envolve leitura ou gravação no banco de dados:

• sqlalchemy.exc.PendingRollbackError
• sqlalchemy.exc.InvalidRequestError

sqlalchemy.exc.PendingRollbackError

Esse erro indica que há uma transação não confirmada em andamento que precisa ser revertida antes de prosseguir com qualquer outra operação do banco de dados.

A causa mais comum desse erro é uma exceção não tratada que ocorre durante uma transação do banco de dados, o que impede que a transação seja confirmada ou revertida corretamente.

sqlalchemy.exc.InvalidRequestError

Esse erro indica que uma operação ou solicitação feita ao banco de dados é inválida ou não suportada. Pode haver várias causas para esse erro, incluindo:

• Uso incorreto da API SQLAlchemy, como chamar um método ou acessar um atributo que não existe ou não é aplicável no contexto fornecido.

• Sintaxe SQL ou estrutura de consulta incorreta.

• Mapeamento ausente ou incorreto entre objetos Python e tabelas/colunas do banco de dados.

Tenho certeza de que você tem ideias sobre como esses erros podem ser resolvidos, no entanto, gostaria de afirmar que, embora eu tenha identificado qual era o problema e feito uma correção, o problema persiste.

Se você está curioso para saber como eu soluciono problemas e os resolvo, considere seguir as seguintes etapas:

sqlalchemy.exc.PendingRollbackError:

• Revise seu código para quaisquer exceções não tratadas que possam impedir que as transações sejam confirmadas ou revertidas corretamente. Assegure-se de que você esteja lidando adequadamente com as exceções e confirmando ou revertendo a transação conforme necessário.

• Verifique se há transações de execução longa ou aninhadas que possam levar a reversões pendentes. Certifique-se de que todas as transações sejam confirmadas ou revertidas em tempo hábil.

• Revise seu código para todos os casos em que você pode estar iniciando uma nova transação antes de lidar adequadamente com a transação anterior.

sqlalchemy.exc.InvalidRequestError:

• Revise a mensagem de erro específica fornecida com a exceção para identificar a causa da solicitação inválida. Isso pode fornecer dicas sobre qual parte do seu código ou instrução SQL está causando o problema.

• Verifique seu código SQLAlchemy para quaisquer chamadas de método incorretas, acessos de atributo ou uso incorreto da API.

• Revise suas instruções SQL para garantir que elas tenham a sintaxe e a estrutura corretas.

• Verifique se o esquema do banco de dados e os mapeamentos SQLAlchemy estão sincronizados. Certifique-se de que todas as tabelas e colunas necessárias existam e que seus objetos Python sejam mapeados adequadamente para as entidades de banco de dados correspondentes.

Vamos ver como trabalhei em uma solução permanente que provou funcionar para mim. Vou continuar usando um projeto no qual trabalhei quando estava demonstrando como usar padrões de repositório.

Tínhamos um módulo onde armazenávamos nosso mixin de sessão base orm com os seguintes códigos:

 # SQLAlchemy Imports from sqlalchemy.orm import Session # Own Imports from config.database import SessionLocal from core.settings import ledger_settings class ORMSessionMixin: """Base orm session mixin for interacting with the database.""" def __init__(self): """ Get the next database session from the database pool. """ self.orm: Session = self.get_db().__next__() def get_db(self): """ This method creates a database session, yields it, rollback the transaction if there's an exception and then finally closes the session. Yields: db: scoped database session """ db = SessionLocal() try: yield db except Exception: db.rollback() finally: db.close()

O problema com esta solução era que, se ocorresse uma exceção no processo de uma transação (pode ser qualquer coisa: criar um usuário, financiar sua carteira, etc.) - as exceções não são tratadas adequadamente e a sessão do banco de dados em trânsito não obter reversão.

Após três meses de depuração e aplicação de patches e muita pesquisa, finalmente consegui criar uma maneira eficiente de lidar com as sessões.

 # SQLAlchemy Imports import sqlalchemy from sqlalchemy.orm import Session # Own Imports from config.database.connection import SessionLocal class DatabaseSessionMixin: """Database session mixin.""" def __enter__(self) -> Session: self.db = SessionLocal() return self.db def __exit__(self, exc_type, exc_val, exc_tb): try: if exc_type is not None: self.db.rollback() except sqlalchemy.exc.SQLAlchemyError: pass finally: self.db.close() SessionLocal.remove() def use_database_session(): return DatabaseSessionMixin()

Neste código:

• DatabaseSession é uma classe gerenciadora de contexto que manipula a sessão e garante que ela seja fechada corretamente e revertida em caso de erro.

• O método __enter__ inicializa a sessão e a retorna.

• O método __exit__ verifica as exceções e reverte a sessão se ocorrer uma exceção. Em seguida, ele fecha a sessão e a remove da sessão com escopo.

• use_database_session é uma função utilitária que pode ser usada como decorador ou gerenciador de contexto para simplificar o uso da sessão.

Aqui está um exemplo de como você pode usar a função utilitária use_database_session:

 with use_database_session() as db: # perform logic that uses the session # ... # After exiting the context, the session will be automatically closed and removed from the scoped session.

A abordagem acima fornece uma maneira mais limpa e eficiente de lidar com sessões e garante que elas sejam revertidas ou fechadas adequadamente em caso de erro. Vamos prosseguir para como você implementa o padrão de repositório ao usar a sessão de banco de dados no ORM.

 # SQLAlchemy Imports import sqlalchemy from sqlalchemy.orm import Session class BaseRepository: def __init__(self, session: Session): self.db = session class UserRepository(BaseRepository): """Operations to interact with the `users` table in the database.""" def get(self, user_id: int) -> User: """This method gets a user from the database.""" user = ( self.db.query(User) .filter(User.id == user_id) .first() ) return user def create(self, name: str, email: str, password: str) -> User: """This method creates a user.""" user = User(name=name, email=email, password=password) self.db.add(user) self.db.commit() self.db.refresh(user) return user def update_user(self, user_id: int, updated_data: dict): """This method updates a user.""" user = self.get(user_id) if user: for key, value in updated_data.items(): setattr(user, key, value) self.db.commit() return user return None def delete_user(self, user_id): """This method deletes a user.""" user = self.get_user(user_id) if user: self.db.delete(user) self.db.commit() return True return False

Em seguida, seria integrar o repositório acima na camada de serviço do seu aplicativo. Suponha que você tenha uma função de serviço que crie contas de usuários; veja como você faria isso usando nosso novo método:

 # Apps Imports from apps.users.models import User from apps.users.repo import UserRepository from apps.users.schemas.auth import UserCreate # Config Imports from config.security.hashers import password from config.database.session_mixin import use_database_session async def create_user(user: UserCreate) -> User: """ This function creates a new user in the database. :param user: schemas.UserCreate :type user: schemas.UserCreate :return: The user object """ with use_database_session() as db: users_repo = UserRepository(db) user = users_repo.create( user.name, user.email, password.hash(user.password) ) return user

O padrão acima permitirá encapsular operações de banco de dados dentro de classes de repositório enquanto aproveita a sessão de banco de dados herdada. Ele também fornece uma separação clara entre seus modelos ORM e a lógica do repositório.

Conclusão

Em conclusão, a manipulação eficiente de sessões é importante ao construir sistemas de back-end.

Os erros como sqlalchemy.exc.PendingRollbackError e sqlalchemy.exc.InvalidRequestError que ocorrem durante transações de banco de dados podem levar a inconsistências de dados e falhas de aplicativos se não forem tratados adequadamente.

Identificar e resolver esses erros é importante para manter a integridade e a confiabilidade do sistema.

Para resolver os problemas relacionados ao tratamento de sessão, é essencial implementar estratégias robustas. Uma abordagem é usar gerenciadores de contexto, como o DatabaseSessionMixin que demonstramos no artigo.

Esse gerenciador de contexto garante que as sessões sejam abertas, fechadas e revertidas adequadamente em caso de exceções. Ao encapsular a lógica da sessão no gerenciador de contexto, você pode simplificar o gerenciamento de sessão e melhorar o tratamento de erros.

Além disso, integrar o padrão de repositório na camada de serviço do aplicativo pode aumentar ainda mais a eficiência do tratamento da sessão.

Ao separar as operações do banco de dados em classes de repositório e aproveitar a sessão herdada do gerenciador de contexto, você pode obter uma organização de código mais limpa e manter uma separação clara entre os modelos ORM e a lógica do repositório.

No geral, a manipulação eficiente da sessão é crucial para manter a consistência dos dados, evitar erros e garantir a estabilidade dos sistemas de back-end.

Ao seguir as melhores práticas, como usar gerenciadores de contexto e adotar o padrão de repositório, os desenvolvedores podem criar sistemas robustos e confiáveis que gerenciam sessões com eficiência e lidam com erros durante as transações do banco de dados.

Estou aberto para escrever shows e procurar ativamente por papéis de contrato que envolvam a construção com Python (Django, FastAPI, etc.).