Garantindo atomicidade de operações com Unit of Work (UoW) Pattern em Python com MongoDB

No desenvolvimento de software, a integridade e a consistência dos dados são fundamentais, especialmente quando lidamos com múltiplas operações de banco de dados que precisam ser tratadas como uma única transação. É aqui que o padrão Unit of Work (UoW) se destaca como uma solução eficaz.

O padrão Unit of Work é uma abordagem poderosa que nos permite agrupar várias operações em uma única transação, garantindo que todas elas sejam concluídas com sucesso ou nenhuma delas seja aplicada. Isso é essencial para garantir as propriedades ACID (Atomicidade, Consistência, Isolamento e Durabilidade) das transações, proporcionando uma maneira robusta de manter a integridade dos dados.

Neste artigo, exploraremos como implementar o padrão Unit of Work em Python utilizando o MongoDB como exemplo de banco de dados. Embora utilizemos Python e MongoDB, os conceitos discutidos podem ser aplicados a outras linguagens e sistemas de banco de dados. Vamos abordar a importância da atomicidade nas operações de banco de dados e como o padrão Unit of Work pode nos ajudar a alcançar isso de maneira eficiente e eficaz.

O que é o padrão Unit of Work?

O padrão Unit of Work é um padrão de projeto que age como um intermediário entre a aplicação e o banco de dados, gerenciando automaticamente o início, o commit e o rollback das transações. Ele rastreia todas as alterações que são feitas durante uma transação e garante que todas elas sejam aplicadas de forma atômica. Se qualquer parte da transação falhar, todas as alterações podem ser revertidas, mantendo a integridade dos dados.

As principais vantagens do padrão Unit of Work incluem:

• Garantia de atomicidade das transações
• Redução de chamadas ao banco de dados, agrupando operações
• Facilitação do gerenciamento de transações complexas
• Melhoria da manutenção e clareza do código

Além disso, este padrão pode ser utilizado para outros recursos, além do banco de dados. Por exemplo, suponha que você tenha que persistir o dado e publica-lo em um tópico ou em uma fila, você poderia garantir a consistência e atomicidade para este caso também.

A importância das propriedades ACID

As propriedades ACID (Atomicidade, Consistência, Isolamento e Durabilidade) são fundamentais para garantir a confiabilidade das transações em um sistema de banco de dados.

• Atomicidade: Garante que todas as operações dentro de uma transação sejam concluídas com sucesso. Se uma operação falhar, todas as outras operações são revertidas.
• Consistência: Assegura que uma transação leva o banco de dados de um estado consistente para outro estado consistente, mantendo as regras e restrições definidas.
• Isolamento: Garante que as transações ocorrem independentemente umas das outras, prevenindo interferências.
• Durabilidade: Assegura que, uma vez confirmada, uma transação permanece permanente, mesmo no caso de uma falha de sistema.

Manter essas propriedades é essencial para evitar problemas como perda de dados, corrupção de dados e inconsistências.

Implementando o padrão Unit of Work com Python e MongoDB

Configuração do ambiente

Antes de começar, vamos configurar o nosso ambiente de desenvolvimento. Para realizar o desenvolvimento, você deve ter o Python, o Docker e o Docker Compose previamente instalados.

Instale as dependências necessárias:

pip install motor pydantic

Definindo as entidades e classes abstratas

Para demonstração, vamos criar uma simples classe de usuário e um Repository para definir quais operações poderão ser realizadas com uma fonte de dados:

from pydantic import BaseModel

from abc import ABC, abstractmethod

class User(BaseModel):
  id: str
  username: str
  email: str

class UserRepository(ABC):
    @abstractmethod
    async def get(user_id: str, **kwargs) -> User | None:
        """Get user by id."""
        ...

    @abstractmethod
    async def create(user: User, **kwargs) -> None:
        """Create new user."""
        ...

    @abstractmethod
    async def update(user: User, **kwargs) -> None:
        """Update an user."""
        ...

    @abstractmethod
    async def delete(user_id: str, **kwargs) -> None:
        """Delete an user."""
        ...

Em seguida, vamos criar uma classe abstrata para representar o padrão UoW, que fornecerá métodos para iniciar, confirmar e reverter transações no banco de dados. Essa classe abstrata nos permitirá criar implementações específicas para diferentes fontes de dados, mantendo a mesma interface para interagir com elas:

from abc import ABC, abstractmethod
from typing import Any

from domain.repositories import UserRepository

class UserUnitOfWork(ABC):
    """User unit of work."""

    users: UserRepository
    session: Any

    async def __aenter__(self) -> "UserUnitOfWork":
        return self

    async def __aexit__(self, *args) -> None:
        await self.rollback()

    @abstractmethod
    async def commit(self) -> None:
        """Commit changes."""
        raise NotImplementedError()

    @abstractmethod
    async def rollback(self) -> None:
        """Rollback changes."""
        raise NotImplementedError()

Antes de prosseguir com a implementação do padrão Unit of Work (UoW), é importante entender o conceito de context manager em Python e como ele se relaciona com os métodos __aenter__ e __aexit__.

Em Python, um context manager é um objeto que define ações a serem executadas antes e depois de um bloco de código específico. Isso é útil para garantir que recursos sejam devidamente gerenciados, como fechamento de arquivos ou conexões de banco de dados.

Os métodos __enter__ e __exit__ são os magic methods especiais que permitem que um objeto se comporte como um context manager. Quando um objeto é usado em um bloco with, o método __enter__ é chamado antes do bloco de código ser executado, e o método __exit__ é chamado depois que o bloco é concluído, independentemente de ocorrer uma exceção ou não.

No entanto, quando estamos lidando com código assíncrono em Python, como é o caso com o uso de coroutines e asyncio, usamos os métodos __aenter__ e __aexit__ em vez de __enter__ e __exit__. A letra "a" em frente a "enter" e "exit" indica que esses métodos são assíncronos.

O método __aenter__ é responsável por inicializar e retornar o recurso que será gerenciado pelo context manager. No contexto de uma classe de UoW, isso pode envolver a inicialização de uma transação de banco de dados.

O método __aexit__ é chamado após o bloco de código ser executado, e é onde normalmente são feitas operações de confirmação ou reversão de transações, dependendo do resultado da operação dentro do bloco with.

Por fim, para utilizarmos os métodos da classe UserUnitOfWork e UserRepository, podemos criar um novo arquivo user_usecases.py para definirmos alguns casos de uso da nossa aplicação:

from domain.errors import UserNotFoundError
from domain.models import User
from domain.uow import UserUnitOfWork

class UserNotFoundError(Exception): ...

async def create_user(user: User, uow: UserUnitOfWork) -> None:
    """Create new user."""
    async with uow:
        await uow.users.create(user, uow.session)

        await uow.commit()

async def update_user(user: User, uow: UserUnitOfWork) -> None:
    """Update an user."""
    async with uow:
        if not await uow.users.get(user.id):
            raise UserNotFoundError()

        await uow.users.update(user, uow.session)

        await uow.commit()

async def delete_user(user_id: str, uow: UserUnitOfWork) -> None:
    """Delete an user."""
    async with uow:
        if not await uow.users.get(user_id):
            raise UserNotFoundError()

        await uow.users.delete(user_id, uow.session)

        await uow.commit()

async def get_user(user_id: str, uow: UserUnitOfWork) -> User | None:
    """Get user by id."""
    return await uow.users.get(user_id)

Implementação das classes abstratas

Agora que definimos nossas entidades de domínio e repositórios, é hora de criar as implementações concretas desses repositórios para interagir com o MongoDB. Utilizaremos a biblioteca motor para conectar e interagir de forma assíncrona com o MongoDB.

Primeiro, vamos criar a implementação do repositório de usuários, UserRepositoryMongo. Esta classe será responsável por realizar operações CRUD no MongoDB para usuários.

from typing import Any
from domain.models import User
from domain.repositories import UserRepository
from motor.motor_asyncio import AsyncIOMotorClient
from motor.core import (
    AgnosticClient,
    AgnosticDatabase,
    AgnosticCollection,
    AgnosticClientSession,
)

class UserRepositoryMongo(UserRepository):
    connection: AgnosticClient

    def __init__(self, connection_string: str) -> None:
        self.connection: AgnosticClient = AsyncIOMotorClient(connection_string)

    def _get_db(self) -> AgnosticDatabase:
        return self.connection.get_database("test")

    def _get_collection(self, collection: str) -> AgnosticCollection:
        return self._get_db().get_collection(collection)

    async def get(self, user_id: str, session: Any = None) -> User | None:
        user = await self._get_collection("users").find_one({"id": user_id})
        if not user:
            return None
        return User(**user)

    async def create(self, user: User, session: AgnosticClientSession) -> None:
        await self._get_collection("users").insert_one(
            user.model_dump(), session=session
        )

    async def update(self, user: User, session: AgnosticClientSession) -> None:
        await self._get_collection("users").update_one(
            {"id": user.id}, {"$set": user.model_dump()}, session=session
        )

    async def delete(self, user_id: str, session: AgnosticClientSession) -> None:
        await self._get_collection("users").delete_one({"id": user_id}, session=session)

A seguir, vamos implementar a classe UserUnitOfWorkMongo, que representa a unidade de trabalho para operações relacionadas a usuários definidas no contrato da classe UserUnitOfWork:

from domain.uow import UserUnitOfWork

from motor.core import AgnosticClientSession

from repositories.mongo_repository import UserRepositoryMongo

class UserUnitOfWorkMongo(UserUnitOfWork):
    def __init__(
        self, session_factory: AgnosticClientSession, user_repo: UserRepositoryMongo
    ):
        self.session_factory = session_factory
        self.users = user_repo
        self.session: AgnosticClientSession | None = None

    async def __aenter__(self) -> "UserUnitOfWorkMongo":
        self.session = await self.users.connection.start_session()
        self.session.start_transaction()
        return self

    async def __aexit__(self, exc_type, *_) -> None:
        if exc_type:
            await self.rollback()
        else:
            await self.commit()
        await self.session.end_session()

    async def commit(self) -> None:
        await self.session.commit_transaction()

    async def rollback(self) -> None:
        await self.session.abort_transaction()

Com essas implementações, estaremos prontos para realizar operações assíncronas no MongoDB de forma eficiente e segura, garantindo a integridade dos dados em transações.

Testando a Implementação com o MongoDB

Agora que implementamos nossos repositórios e unidades de trabalho para interagir com o MongoDB de forma assíncrona, vamos testar nossa implementação com um exemplo simples.

Primeiro, vamos configurar um arquivo docker-compose.yaml para criarmos um container MongoDB sendo executado como replica-set com nó único, pois transações do MongoDB são permitidas somente em replica-sets ou shared clusters:

version: "3.8"

services:
  mongo1:
    image: mongo:7.0
    command: ["--replSet", "rs0", "--bind_ip_all", "--port", "27017"]
    ports:
      - 27017:27017
    volumes:
      - "mongo_data:/data/db"
      - "mongo_config:/data/configdb"

volumes:
  mongo_data:
  mongo_config:

No código abaixo, vamos criar um usuário, persisti-lo no banco de dados e em seguida recuperá-lo. Isso nos permitirá verificar se as operações de criação e recuperação de usuário estão funcionando corretamente.

import asyncio

from domain.models import User
from domain.usecases import create_user, get_user
from repositories.mongo_repository import UserRepositoryMongo
from uow.uow_mongo import UserUnitOfWorkMongo
from motor.motor_asyncio import AsyncIOMotorClientSession

async def uow_factory() -> UserUnitOfWorkMongo:
    repo = UserRepositoryMongo("mongodb://localhost:27017")

    session = await repo.connection.start_session()

    session = AsyncIOMotorClientSession(
        delegate=session,
        motor_client=repo.connection,
    )

    return UserUnitOfWorkMongo(
        session_factory=session,
        user_repo=repo,
    )

async def create_and_get_user() -> None:
    user = User(id="test", name="John", email="johndoe")
    uow = await uow_factory()

    await create_user(user, uow)

    user = await get_user("test", uow)

    print(user)

if __name__ == "__main__":
    asyncio.run(create_and_get_user())

> python main.py

id='test' name='John' email='johndoe'

Conclusão

Em resumo, neste artigo exploramos a implementação do padrão Unit of Work (UoW) em Python com o MongoDB. Começamos definindo as entidades e repositórios necessários para nossa aplicação, em seguida, construímos uma classe abstrata para representar nossa unidade de trabalho, que encapsula as operações de transação.

Demonstramos como criar um UoW concreto que utiliza o MongoDB como fonte de dados, juntamente com repositórios MongoDB específicos. Além disso, explicamos a importância dos context managers e dos métodos __aenter__ e __aexit__ na implementação assíncrona do Python.

Por fim, testamos nossa implementação com um exemplo prático, mostrando como criar e recuperar usuários de um banco de dados MongoDB em um ambiente assíncrono.

Esperamos que este artigo tenha sido útil para você compreender como utilizar o padrão UoW em conjunto com o MongoDB em suas aplicações Python. Ao incorporar esse padrão em seus projetos, você estará promovendo uma arquitetura mais robusta, modular e de fácil manutenção.

Se você tiver alguma dúvida ou feedback, não hesite em entrar em contato. Obrigado por ler e boa sorte em seus projetos de desenvolvimento!

Referências:

• https://medium.com/workleap/the-only-local-mongodb-replica-set-with-docker-compose-guide-youll-ever-need-2f0b74dd8384
• https://www.cosmicpython.com/book/chapter_02_repository.html
• https://www.cosmicpython.com/book/chapter_06_uow.html