Repositories

Camada anticorrupção: traduz exceções técnicas em erros de domínio.

py_return_success_or_error.repositories.repository

Contrato da camada de repositório.

Repository

Bases: ABC

Contrato do repositório: nunca lança falha técnica ao chamador.

É a abstração que o caso de uso recebe (inversão de dependência): devolve sempre Success[TData] ou Failure[TError], com o erro já traduzido para o conjunto fechado da feature.

Source code in src/py_return_success_or_error/repositories/repository.py

class Repository[TData, TParams: Parameters, TError](ABC):
    """Contrato do repositório: nunca lança falha técnica ao chamador.

    É a abstração que o caso de uso recebe (inversão de dependência):
    devolve sempre ``Success[TData]`` ou ``Failure[TError]``, com o erro
    já traduzido para o conjunto fechado da feature.
    """

    @abstractmethod
    async def __call__(
        self, parameters: TParams
    ) -> ReturnSuccessOrError[TData, TError]:
        """Busca o dado e traduz falhas técnicas em erro de domínio.

        Args:
            parameters: Parâmetros da busca.

        Returns:
            ``Success`` com o dado, ou ``Failure`` com o erro da feature.
        """
        ...  # pragma: no cover

__call__(parameters) abstractmethod async

Busca o dado e traduz falhas técnicas em erro de domínio.

Parameters:

Name	Type	Description	Default
parameters	TParams	Parâmetros da busca.	required

Returns:

Type	Description
ReturnSuccessOrError[TData, TError]	Success com o dado, ou Failure com o erro da feature.

Source code in src/py_return_success_or_error/repositories/repository.py

@abstractmethod
async def __call__(
    self, parameters: TParams
) -> ReturnSuccessOrError[TData, TError]:
    """Busca o dado e traduz falhas técnicas em erro de domínio.

    Args:
        parameters: Parâmetros da busca.

    Returns:
        ``Success`` com o dado, ou ``Failure`` com o erro da feature.
    """
    ...  # pragma: no cover

py_return_success_or_error.repositories.repository_base

Implementação base do repositório — a fronteira try/except.

RepositoryBase

Bases: Repository[TData, TParams, TError]

Camada anticorrupção sobre um :class:DataSource.

Envolve a chamada ao datasource numa fronteira try/except:

• sucesso → Success[TData];
• exceção técnica → :meth:map_error → Failure[TError];
• asyncio.CancelledError → propaga (cancelamento não é falha de domínio).

Source code in src/py_return_success_or_error/repositories/repository_base.py

class RepositoryBase[TData, TParams: Parameters, TError](
    Repository[TData, TParams, TError]
):
    """Camada anticorrupção sobre um :class:`DataSource`.

    Envolve a chamada ao datasource numa fronteira try/except:

    - sucesso → ``Success[TData]``;
    - exceção técnica → :meth:`map_error` → ``Failure[TError]``;
    - ``asyncio.CancelledError`` → **propaga** (cancelamento não é falha
      de domínio).
    """

    def __init__(self, datasource: DataSource[TData, TParams]) -> None:
        """Inicializa com a porta de dados a proteger.

        Args:
            datasource: A porta "burra" de acesso ao recurso externo.
        """
        self._datasource = datasource

    async def __call__(
        self, parameters: TParams
    ) -> ReturnSuccessOrError[TData, TError]:
        """Executa o datasource dentro da fronteira de erro.

        Args:
            parameters: Parâmetros da busca.

        Returns:
            ``Success`` com o dado bruto, ou ``Failure`` com o erro
            traduzido por :meth:`map_error`.
        """
        try:
            return Success(await self._datasource(parameters))
        except asyncio.CancelledError:
            raise
        except Exception as exception:
            return Failure(self.map_error(exception, parameters))

    @abstractmethod
    def map_error(self, exception: Exception, parameters: TParams) -> TError:
        """Traduz a exceção técnica num caso do conjunto fechado da feature.

        **Abstrato**: não há erro universal a fabricar — cada repositório
        decide como cada exceção vira um caso de ``TError`` (tipicamente
        com um caso "genérico" para o inesperado).

        Args:
            exception: A exceção técnica capturada.
            parameters: Os parâmetros da chamada que falhou.

        Returns:
            O erro de domínio correspondente.
        """
        ...  # pragma: no cover

__call__(parameters) async

Executa o datasource dentro da fronteira de erro.

Parameters:

Name	Type	Description	Default
parameters	TParams	Parâmetros da busca.	required

Returns:

Type	Description
ReturnSuccessOrError[TData, TError]	Success com o dado bruto, ou Failure com o erro
ReturnSuccessOrError[TData, TError]	traduzido por :meth:map_error.

Source code in src/py_return_success_or_error/repositories/repository_base.py

async def __call__(
    self, parameters: TParams
) -> ReturnSuccessOrError[TData, TError]:
    """Executa o datasource dentro da fronteira de erro.

    Args:
        parameters: Parâmetros da busca.

    Returns:
        ``Success`` com o dado bruto, ou ``Failure`` com o erro
        traduzido por :meth:`map_error`.
    """
    try:
        return Success(await self._datasource(parameters))
    except asyncio.CancelledError:
        raise
    except Exception as exception:
        return Failure(self.map_error(exception, parameters))

__init__(datasource)

Inicializa com a porta de dados a proteger.

Parameters:

Name	Type	Description	Default
datasource	DataSource[TData, TParams]	A porta "burra" de acesso ao recurso externo.	required

Source code in src/py_return_success_or_error/repositories/repository_base.py

def __init__(self, datasource: DataSource[TData, TParams]) -> None:
    """Inicializa com a porta de dados a proteger.

    Args:
        datasource: A porta "burra" de acesso ao recurso externo.
    """
    self._datasource = datasource

map_error(exception, parameters) abstractmethod

Traduz a exceção técnica num caso do conjunto fechado da feature.

Abstrato: não há erro universal a fabricar — cada repositório decide como cada exceção vira um caso de TError (tipicamente com um caso "genérico" para o inesperado).

Parameters:

Name	Type	Description	Default
exception	Exception	A exceção técnica capturada.	required
parameters	TParams	Os parâmetros da chamada que falhou.	required

Returns:

Type	Description
TError	O erro de domínio correspondente.

Source code in src/py_return_success_or_error/repositories/repository_base.py

@abstractmethod
def map_error(self, exception: Exception, parameters: TParams) -> TError:
    """Traduz a exceção técnica num caso do conjunto fechado da feature.

    **Abstrato**: não há erro universal a fabricar — cada repositório
    decide como cada exceção vira um caso de ``TError`` (tipicamente
    com um caso "genérico" para o inesperado).

    Args:
        exception: A exceção técnica capturada.
        parameters: Os parâmetros da chamada que falhou.

    Returns:
        O erro de domínio correspondente.
    """
    ...  # pragma: no cover