Pular para conteúdo

Usecases

Orquestração, regra de negócio e fronteira do inesperado.

py_return_success_or_error.usecases.usecase_executor_base

Base compartilhada pelos casos de uso.

UsecaseExecutorBase

Bases: ABC

Base compartilhada pelos casos de uso.

Concentra o que é comum aos dois fluxos (:class:UsecaseBase e :class:UsecaseBaseCallData): a medição opcional do tempo e o despacho opcional do processamento (CPU-bound) a uma thread. As subclasses definem o fluxo específico e implementam o process.

Attributes:

Name Type Description
run_in_background

Se verdadeiro, o process (CPU-bound) roda fora do event loop via :meth:_dispatch_to_background (padrão: asyncio.to_thread), mantendo o loop responsivo. No build padrão do CPython o GIL limita o paralelismo de CPU puro (no free-threaded, 3.14+, o paralelismo é real); uma thread já iniciada não é interrompida pelo cancelamento da task.
monitor_execution_time

Se verdadeiro, mede o tempo de execução e o entrega a :meth:on_execution_time_measured.
Source code in src/py_return_success_or_error/usecases/usecase_executor_base.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
class UsecaseExecutorBase[TValue, TError](ABC):
    """Base compartilhada pelos casos de uso.

    Concentra o que é comum aos dois fluxos (:class:`UsecaseBase` e
    :class:`UsecaseBaseCallData`): a medição opcional do tempo e o
    despacho opcional do processamento (CPU-bound) a uma thread. As
    subclasses definem o fluxo específico e implementam o ``process``.

    Attributes:
        run_in_background: Se verdadeiro, o ``process`` (CPU-bound) roda
            fora do event loop via :meth:`_dispatch_to_background`
            (padrão: ``asyncio.to_thread``), mantendo o loop responsivo. No build padrão do CPython o
            GIL limita o paralelismo de CPU puro (no free-threaded,
            3.14+, o paralelismo é real); uma thread já iniciada não é
            interrompida pelo cancelamento da task.
        monitor_execution_time: Se verdadeiro, mede o tempo de execução
            e o entrega a :meth:`on_execution_time_measured`.
    """

    def __init__(
        self,
        *,
        run_in_background: bool = False,
        monitor_execution_time: bool = False,
    ) -> None:
        """Configura as flags de execução do caso de uso.

        Args:
            run_in_background: Despacha o ``process`` para uma thread.
            monitor_execution_time: Habilita a medição de tempo.
        """
        self.run_in_background = run_in_background
        self.monitor_execution_time = monitor_execution_time

    @abstractmethod
    def on_unexpected(self, exception: Exception) -> TError:
        """Converte uma exceção **inesperada** do ``process`` (um bug)
        num erro do conjunto fechado da feature.

        **Abstrato**: como não há erro universal a fabricar, o caso de
        uso decide para qual caso de ``TError`` o inesperado é mapeado
        (tipicamente um caso "Unexpected"/``ErrorGeneric``). Garante que
        o ``process`` nunca lança ao chamador — o resultado é sempre um
        dos casos previstos.

        **Exceção do contrato — cancelamento**: um
        ``asyncio.CancelledError`` não passa por aqui: cancelamento não é
        falha de domínio e propaga como exceção, no idioma do asyncio.

        Args:
            exception: A exceção inesperada capturada.

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

    def on_execution_time_measured(self, elapsed: timedelta) -> None:
        """Recebe o tempo medido quando ``monitor_execution_time`` está
        habilitado.

        **Virtual**: sobrescreva para integrar à sua observabilidade —
        a base não impõe dependência de logging. A implementação padrão
        registra em ``logging`` no nível DEBUG (logger
        ``py_return_success_or_error``).

        Args:
            elapsed: Duração total da execução do caso de uso.
        """
        modo = 'Background' if self.run_in_background else 'Direct'
        _logger.debug(
            '[py-return-success-or-error] Execution Time %s (%s): %.2fms',
            type(self).__name__,
            modo,
            elapsed.total_seconds() * 1000,
        )

    def ok(self, value: TValue) -> ReturnSuccessOrError[TValue, TError]:
        """Cria um resultado de **sucesso** a partir do valor.

        Conveniência simétrica a :meth:`fail`: ``self.ok(valor)`` é a
        forma recomendada de retornar sucesso dentro do ``process``.

        Args:
            value: O valor de sucesso.

        Returns:
            ``Success(value)`` tipado com os parâmetros do caso de uso.
        """
        return Success(value)

    def fail(self, error: TError) -> ReturnSuccessOrError[TValue, TError]:
        """Cria um resultado de **falha** a partir de um caso concreto do
        conjunto fechado de erro.

        É a forma recomendada de retornar um erro de negócio no
        ``process`` — resolve a inferência de tipos do resultado.

        Args:
            error: Um caso concreto de ``TError``.

        Returns:
            ``Failure(error)`` tipado com os parâmetros do caso de uso.
        """
        return Failure(error)

    async def _measured(
        self,
        run: Callable[[], Awaitable[ReturnSuccessOrError[TValue, TError]]],
    ) -> ReturnSuccessOrError[TValue, TError]:
        """Envolve a execução com a medição de tempo, quando habilitada."""
        if not self.monitor_execution_time:
            return await run()

        start = time.perf_counter()
        result = await run()
        elapsed = timedelta(seconds=time.perf_counter() - start)
        self.on_execution_time_measured(elapsed)
        return result

    async def _dispatch_to_background(
        self,
        run: Callable[[], ReturnSuccessOrError[TValue, TError]],
    ) -> ReturnSuccessOrError[TValue, TError]:
        """Despacha o processamento quando ``run_in_background`` está
        habilitado.

        **Virtual**: o padrão é ``asyncio.to_thread``, que mantém o
        event loop responsivo. No CPython **free-threaded**
        (3.14+, PEP 779) isso já dá paralelismo real de CPU; no build
        padrão, o GIL limita o paralelismo de CPU puro — sobrescreva
        para usar outro executor quando precisar, por exemplo o
        ``InterpreterPoolExecutor`` do Python 3.14+ (PEP 734)::

            async def _dispatch_to_background(self, run):
                loop = asyncio.get_running_loop()
                return await loop.run_in_executor(self._executor, run)

        Args:
            run: O processamento já protegido (nunca lança, exceto
                cancelamento).

        Returns:
            O resultado do processamento.
        """
        return await asyncio.to_thread(run)

    async def _process_stage(
        self,
        process: Callable[[], ReturnSuccessOrError[TValue, TError]],
    ) -> ReturnSuccessOrError[TValue, TError]:
        """Executa o ``process`` direto ou, se ``run_in_background``,
        via :meth:`_dispatch_to_background`.

        Em **ambos** os modos, uma exceção inesperada é convertida via
        :meth:`on_unexpected` em ``Failure`` — o ``process`` nunca
        propaga exceção ao chamador. Única exceção: o **cancelamento**
        (``asyncio.CancelledError``) propaga em ambos os modos —
        cancelamento não é falha de domínio.
        """
        # Paridade direto↔background: um cancelamento já pendente é
        # entregue ANTES do process, nos dois modos.
        await asyncio.sleep(0)

        def guarded() -> ReturnSuccessOrError[TValue, TError]:
            try:
                return process()
            except asyncio.CancelledError:
                raise
            except Exception as exception:
                return Failure(self.on_unexpected(exception))

        if not self.run_in_background:
            return guarded()

        return await self._dispatch_to_background(guarded)

__init__(*, run_in_background=False, monitor_execution_time=False)

Configura as flags de execução do caso de uso.

Parameters:

Name Type Description Default
run_in_background bool

Despacha o process para uma thread.

 False
monitor_execution_time bool

Habilita a medição de tempo.

 False
Source code in src/py_return_success_or_error/usecases/usecase_executor_base.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(
    self,
    *,
    run_in_background: bool = False,
    monitor_execution_time: bool = False,
) -> None:
    """Configura as flags de execução do caso de uso.

    Args:
        run_in_background: Despacha o ``process`` para uma thread.
        monitor_execution_time: Habilita a medição de tempo.
    """
    self.run_in_background = run_in_background
    self.monitor_execution_time = monitor_execution_time

fail(error)

Cria um resultado de falha a partir de um caso concreto do conjunto fechado de erro.

É a forma recomendada de retornar um erro de negócio no process — resolve a inferência de tipos do resultado.

Parameters:

Name Type Description Default
error TError

Um caso concreto de TError.

 required

Returns:

Type Description
ReturnSuccessOrError[TValue, TError]

Failure(error) tipado com os parâmetros do caso de uso.
Source code in src/py_return_success_or_error/usecases/usecase_executor_base.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def fail(self, error: TError) -> ReturnSuccessOrError[TValue, TError]:
    """Cria um resultado de **falha** a partir de um caso concreto do
    conjunto fechado de erro.

    É a forma recomendada de retornar um erro de negócio no
    ``process`` — resolve a inferência de tipos do resultado.

    Args:
        error: Um caso concreto de ``TError``.

    Returns:
        ``Failure(error)`` tipado com os parâmetros do caso de uso.
    """
    return Failure(error)

ok(value)

Cria um resultado de sucesso a partir do valor.

Conveniência simétrica a :meth:fail: self.ok(valor) é a forma recomendada de retornar sucesso dentro do process.

Parameters:

Name Type Description Default
value TValue

O valor de sucesso.

 required

Returns:

Type Description
ReturnSuccessOrError[TValue, TError]

Success(value) tipado com os parâmetros do caso de uso.
Source code in src/py_return_success_or_error/usecases/usecase_executor_base.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def ok(self, value: TValue) -> ReturnSuccessOrError[TValue, TError]:
    """Cria um resultado de **sucesso** a partir do valor.

    Conveniência simétrica a :meth:`fail`: ``self.ok(valor)`` é a
    forma recomendada de retornar sucesso dentro do ``process``.

    Args:
        value: O valor de sucesso.

    Returns:
        ``Success(value)`` tipado com os parâmetros do caso de uso.
    """
    return Success(value)

on_execution_time_measured(elapsed)

Recebe o tempo medido quando monitor_execution_time está habilitado.

Virtual: sobrescreva para integrar à sua observabilidade — a base não impõe dependência de logging. A implementação padrão registra em logging no nível DEBUG (logger py_return_success_or_error).

Parameters:

Name Type Description Default
elapsed timedelta

Duração total da execução do caso de uso.

 required
Source code in src/py_return_success_or_error/usecases/usecase_executor_base.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def on_execution_time_measured(self, elapsed: timedelta) -> None:
    """Recebe o tempo medido quando ``monitor_execution_time`` está
    habilitado.

    **Virtual**: sobrescreva para integrar à sua observabilidade —
    a base não impõe dependência de logging. A implementação padrão
    registra em ``logging`` no nível DEBUG (logger
    ``py_return_success_or_error``).

    Args:
        elapsed: Duração total da execução do caso de uso.
    """
    modo = 'Background' if self.run_in_background else 'Direct'
    _logger.debug(
        '[py-return-success-or-error] Execution Time %s (%s): %.2fms',
        type(self).__name__,
        modo,
        elapsed.total_seconds() * 1000,
    )

on_unexpected(exception) abstractmethod

Converte uma exceção inesperada do process (um bug) num erro do conjunto fechado da feature.

Abstrato: como não há erro universal a fabricar, o caso de uso decide para qual caso de TError o inesperado é mapeado (tipicamente um caso "Unexpected"/ErrorGeneric). Garante que o process nunca lança ao chamador — o resultado é sempre um dos casos previstos.

Exceção do contrato — cancelamento: um asyncio.CancelledError não passa por aqui: cancelamento não é falha de domínio e propaga como exceção, no idioma do asyncio.

Parameters:

Name Type Description Default
exception Exception

A exceção inesperada capturada.

 required

Returns:

Type Description
TError

O erro de domínio correspondente.
Source code in src/py_return_success_or_error/usecases/usecase_executor_base.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@abstractmethod
def on_unexpected(self, exception: Exception) -> TError:
    """Converte uma exceção **inesperada** do ``process`` (um bug)
    num erro do conjunto fechado da feature.

    **Abstrato**: como não há erro universal a fabricar, o caso de
    uso decide para qual caso de ``TError`` o inesperado é mapeado
    (tipicamente um caso "Unexpected"/``ErrorGeneric``). Garante que
    o ``process`` nunca lança ao chamador — o resultado é sempre um
    dos casos previstos.

    **Exceção do contrato — cancelamento**: um
    ``asyncio.CancelledError`` não passa por aqui: cancelamento não é
    falha de domínio e propaga como exceção, no idioma do asyncio.

    Args:
        exception: A exceção inesperada capturada.

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

py_return_success_or_error.usecases.usecase_base

Caso de uso de lógica pura, sem acesso a dados.

UsecaseBase

Bases: UsecaseExecutorBase[TValue, TError]

Caso de uso de lógica pura: sem datasource/repositório.

A subclasse implementa apenas :meth:process (síncrono, CPU-bound) e :meth:on_unexpected. O chamador invoca o caso de uso como uma função assíncrona::

result = await usecase(parameters)
Source code in src/py_return_success_or_error/usecases/usecase_base.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class UsecaseBase[TValue, TParams: Parameters, TError](
    UsecaseExecutorBase[TValue, TError]
):
    """Caso de uso de **lógica pura**: sem datasource/repositório.

    A subclasse implementa apenas :meth:`process` (síncrono, CPU-bound) e
    :meth:`on_unexpected`. O chamador invoca o caso de uso como uma
    função assíncrona::

        result = await usecase(parameters)
    """

    @abstractmethod
    def process(
        self, parameters: TParams
    ) -> ReturnSuccessOrError[TValue, TError]:
        """A regra de negócio da feature — **síncrona** e CPU-bound.

        Retorne com ``self.ok(valor)`` / ``self.fail(erro)``. Não deve
        fazer I/O: dados externos entram pelo fluxo
        :class:`UsecaseBaseCallData`.

        Args:
            parameters: Parâmetros de entrada da feature.

        Returns:
            ``Success`` ou ``Failure`` do conjunto fechado da feature.
        """
        ...  # pragma: no cover

    async def __call__(
        self, parameters: TParams
    ) -> ReturnSuccessOrError[TValue, TError]:
        """Orquestra a execução: medição opcional + despacho do process.

        Args:
            parameters: Parâmetros de entrada da feature.

        Returns:
            O resultado do ``process``, com o inesperado já convertido
            via ``on_unexpected``.
        """
        return await self._measured(
            lambda: self._process_stage(lambda: self.process(parameters))
        )

__call__(parameters) async

Orquestra a execução: medição opcional + despacho do process.

Parameters:

Name Type Description Default
parameters TParams

Parâmetros de entrada da feature.

 required

Returns:

Type Description
ReturnSuccessOrError[TValue, TError]

O resultado do process, com o inesperado já convertido
ReturnSuccessOrError[TValue, TError]

via on_unexpected.
Source code in src/py_return_success_or_error/usecases/usecase_base.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
async def __call__(
    self, parameters: TParams
) -> ReturnSuccessOrError[TValue, TError]:
    """Orquestra a execução: medição opcional + despacho do process.

    Args:
        parameters: Parâmetros de entrada da feature.

    Returns:
        O resultado do ``process``, com o inesperado já convertido
        via ``on_unexpected``.
    """
    return await self._measured(
        lambda: self._process_stage(lambda: self.process(parameters))
    )

process(parameters) abstractmethod

A regra de negócio da feature — síncrona e CPU-bound.

Retorne com self.ok(valor) / self.fail(erro). Não deve fazer I/O: dados externos entram pelo fluxo :class:UsecaseBaseCallData.

Parameters:

Name Type Description Default
parameters TParams

Parâmetros de entrada da feature.

 required

Returns:

Type Description
ReturnSuccessOrError[TValue, TError]

Success ou Failure do conjunto fechado da feature.
Source code in src/py_return_success_or_error/usecases/usecase_base.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@abstractmethod
def process(
    self, parameters: TParams
) -> ReturnSuccessOrError[TValue, TError]:
    """A regra de negócio da feature — **síncrona** e CPU-bound.

    Retorne com ``self.ok(valor)`` / ``self.fail(erro)``. Não deve
    fazer I/O: dados externos entram pelo fluxo
    :class:`UsecaseBaseCallData`.

    Args:
        parameters: Parâmetros de entrada da feature.

    Returns:
        ``Success`` ou ``Failure`` do conjunto fechado da feature.
    """
    ...  # pragma: no cover

py_return_success_or_error.usecases.usecase_base_call_data

Caso de uso com busca de dados.

UsecaseBaseCallData

Bases: UsecaseExecutorBase[TValue, TError]

Caso de uso que busca dados antes de aplicar a regra.

Depende da abstração :class:Repository (inversão de dependência) e orquestra três fases:

  1. FETCH — await repository(parameters);
  2. CURTO-CIRCUITO — um Failure do fetch retorna direto ao chamador, sem executar o process;
  3. PROCESS — a regra de negócio recebe o dado limpo.
Source code in src/py_return_success_or_error/usecases/usecase_base_call_data.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class UsecaseBaseCallData[TValue, TData, TParams: Parameters, TError](
    UsecaseExecutorBase[TValue, TError]
):
    """Caso de uso que **busca dados** antes de aplicar a regra.

    Depende da abstração :class:`Repository` (inversão de dependência) e
    orquestra três fases:

    1. **FETCH** — ``await repository(parameters)``;
    2. **CURTO-CIRCUITO** — um ``Failure`` do fetch retorna direto ao
       chamador, sem executar o ``process``;
    3. **PROCESS** — a regra de negócio recebe o dado limpo.
    """

    def __init__(
        self,
        repository: Repository[TData, TParams, TError],
        *,
        run_in_background: bool = False,
        monitor_execution_time: bool = False,
    ) -> None:
        """Inicializa com o repositório da feature.

        Args:
            repository: A abstração de acesso a dados da feature.
            run_in_background: Despacha o ``process`` para uma thread.
            monitor_execution_time: Habilita a medição de tempo.
        """
        super().__init__(
            run_in_background=run_in_background,
            monitor_execution_time=monitor_execution_time,
        )
        self._repository = repository

    @abstractmethod
    def process(
        self, data: TData, parameters: TParams
    ) -> ReturnSuccessOrError[TValue, TError]:
        """A regra de negócio da feature — **síncrona** e CPU-bound.

        Só é chamada quando o fetch teve sucesso: ``data`` chega limpo.
        Retorne com ``self.ok(valor)`` / ``self.fail(erro)``.

        Args:
            data: O dado obtido pelo repositório.
            parameters: Parâmetros de entrada da feature.

        Returns:
            ``Success`` ou ``Failure`` do conjunto fechado da feature.
        """
        ...  # pragma: no cover

    async def __call__(
        self, parameters: TParams
    ) -> ReturnSuccessOrError[TValue, TError]:
        """Orquestra FETCH → CURTO-CIRCUITO → PROCESS.

        Args:
            parameters: Parâmetros de entrada da feature.

        Returns:
            O ``Failure`` do fetch inalterado, ou o resultado do
            ``process`` com o inesperado já convertido.
        """

        async def run() -> ReturnSuccessOrError[TValue, TError]:
            fetch_result = await self._repository(parameters)
            match fetch_result:
                case Failure() as failure:
                    return failure
                case Success(data):
                    return await self._process_stage(
                        lambda: self.process(data, parameters)
                    )
                case _:  # pragma: no cover - provado pelo mypy
                    assert_never(fetch_result)

        return await self._measured(run)

__call__(parameters) async

Orquestra FETCH → CURTO-CIRCUITO → PROCESS.

Parameters:

Name Type Description Default
parameters TParams

Parâmetros de entrada da feature.

 required

Returns:

Type Description
ReturnSuccessOrError[TValue, TError]

O Failure do fetch inalterado, ou o resultado do
ReturnSuccessOrError[TValue, TError]

process com o inesperado já convertido.
Source code in src/py_return_success_or_error/usecases/usecase_base_call_data.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
async def __call__(
    self, parameters: TParams
) -> ReturnSuccessOrError[TValue, TError]:
    """Orquestra FETCH → CURTO-CIRCUITO → PROCESS.

    Args:
        parameters: Parâmetros de entrada da feature.

    Returns:
        O ``Failure`` do fetch inalterado, ou o resultado do
        ``process`` com o inesperado já convertido.
    """

    async def run() -> ReturnSuccessOrError[TValue, TError]:
        fetch_result = await self._repository(parameters)
        match fetch_result:
            case Failure() as failure:
                return failure
            case Success(data):
                return await self._process_stage(
                    lambda: self.process(data, parameters)
                )
            case _:  # pragma: no cover - provado pelo mypy
                assert_never(fetch_result)

    return await self._measured(run)

__init__(repository, *, run_in_background=False, monitor_execution_time=False)

Inicializa com o repositório da feature.

Parameters:

Name Type Description Default
repository Repository[TData, TParams, TError]

A abstração de acesso a dados da feature.

 required
run_in_background bool

Despacha o process para uma thread.

 False
monitor_execution_time bool

Habilita a medição de tempo.

 False
Source code in src/py_return_success_or_error/usecases/usecase_base_call_data.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    repository: Repository[TData, TParams, TError],
    *,
    run_in_background: bool = False,
    monitor_execution_time: bool = False,
) -> None:
    """Inicializa com o repositório da feature.

    Args:
        repository: A abstração de acesso a dados da feature.
        run_in_background: Despacha o ``process`` para uma thread.
        monitor_execution_time: Habilita a medição de tempo.
    """
    super().__init__(
        run_in_background=run_in_background,
        monitor_execution_time=monitor_execution_time,
    )
    self._repository = repository

process(data, parameters) abstractmethod

A regra de negócio da feature — síncrona e CPU-bound.

Só é chamada quando o fetch teve sucesso: data chega limpo. Retorne com self.ok(valor) / self.fail(erro).

Parameters:

Name Type Description Default
data TData

O dado obtido pelo repositório.

 required
parameters TParams

Parâmetros de entrada da feature.

 required

Returns:

Type Description
ReturnSuccessOrError[TValue, TError]

Success ou Failure do conjunto fechado da feature.
Source code in src/py_return_success_or_error/usecases/usecase_base_call_data.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@abstractmethod
def process(
    self, data: TData, parameters: TParams
) -> ReturnSuccessOrError[TValue, TError]:
    """A regra de negócio da feature — **síncrona** e CPU-bound.

    Só é chamada quando o fetch teve sucesso: ``data`` chega limpo.
    Retorne com ``self.ok(valor)`` / ``self.fail(erro)``.

    Args:
        data: O dado obtido pelo repositório.
        parameters: Parâmetros de entrada da feature.

    Returns:
        ``Success`` ou ``Failure`` do conjunto fechado da feature.
    """
    ...  # pragma: no cover