loom.core.use_case

loom.core.use_case.Exists(entity_type, *, from_param=None, from_command=None, against, on_missing=OnMissing.RETURN_FALSE)[source]

Factory returning marker for boolean existence checks.

Parameters:

  • entity_type (type[EntityT])

  • from_param (str | None)

  • from_command (str | None)

  • against (str)

  • on_missing (OnMissing)

Return type:

Any

class loom.core.use_case.Compute[source]

Bases: object

Compute DSL namespace.

Example

Compute.set(F(UpdateUser).slug).from_command(F(UpdateUser).name, via=slugify)

loom.core.use_case.F(root)

Build a typed field-reference factory.

Example

F(UpdateUserCommand).birthdate

Parameters:

root (type[Command] | type[BaseModel] | str)

Return type:

Any

class loom.core.use_case.FieldRef(root, path)[source]

Bases: object

Declarative reference to a field path on a command (or loaded alias).

Parameters:
loom.core.use_case.Input()[source]

Factory returning the runtime marker for command payload parameters.

Returned value is intentionally typed as Any in overloads to avoid mypy default-argument incompatibility in signatures like: cmd: Command = Input().

Return type:

Any

loom.core.use_case.Load(entity_type, *, from_param=None, from_command=None, against, profile='default', on_missing=OnMissing.RAISE)[source]

Factory returning marker for preloaded entity parameters by field.

Parameters:

  • entity_type (type[EntityT])

  • from_param (str | None)

  • from_command (str | None)

  • against (str)

  • profile (str)

  • on_missing (OnMissing)

Return type:

Any

loom.core.use_case.LoadById(entity_type, *, by='id', profile='default', on_missing=OnMissing.RAISE)[source]

Factory returning marker for preloaded entity parameters by id.

Returned value is intentionally typed as Any in overloads to avoid mypy default-argument incompatibility in signatures like: entity: User = LoadById(User, by="id").

Parameters:

  • entity_type (type[EntityT]) – Domain entity type the repository should load.

  • by (str) – Name of the primitive parameter used as the lookup key. Defaults to "id".

  • profile (str) – Loading profile forwarded to repo.get_by_id. Defaults to "default".

  • on_missing (OnMissing) – Missing-entity policy. Defaults to OnMissing.RAISE.

Return type:

Any

class loom.core.use_case.OnMissing(value)[source]

Bases: StrEnum

Policy applied when a marker lookup does not resolve an entity.

class loom.core.use_case.PredicateOp(value)[source]

Bases: StrEnum

class loom.core.use_case.Rule[source]

Bases: object

Rule DSL namespace.

class loom.core.use_case.RuleFn(*args, **kwargs)[source]

Bases: Protocol

class loom.core.use_case.UseCase(main_repo=None)[source]

Bases: ABC, Generic[ModelT, ResultT]

Base class for all use cases.

Subclass and implement execute with typed parameters. Parameter defaults declare the execution contract:

  • Input() — command payload, built from the raw request.

  • LoadById(EntityType, by="param") — entity prefetched by id.

  • Load(EntityType, ...) — entity prefetched by arbitrary field.

  • Exists(EntityType, ...) — boolean existence check by field.

  • No default — primitive param bound directly from the caller.

Class attributes computes, rules, and read_only declare the pre-execution pipeline and execution policy. They are inspected once at startup by UseCaseCompiler and embedded in the immutable ExecutionPlan.

Parameters:

main_repo (RepoFor[Any] | None)

computes

Compute transformations applied in order before rule checks.

Type:

ClassVar[Sequence[ComputeFn[Any]]]

rules

Rule validations applied in order after computes.

Type:

ClassVar[Sequence[RuleFn]]

read_only

When True, the executor skips opening a UnitOfWork transaction. Set this on query-only use cases that never mutate state. GET routes in RestInterface always bypass the UoW regardless of this flag.

Type:

ClassVar[bool]

Example:

class UpdateUserUseCase(UseCase[UserResponse]):
    computes = [set_updated_at]
    rules = [email_must_be_valid]

    def __init__(self, user_repo: UserRepository) -> None:
        self._user_repo = user_repo

    async def execute(
        self,
        user_id: int,
        cmd: UpdateUserCommand = Input(),
        user: User = LoadById(User, by="user_id"),
    ) -> UserResponse:
        ...
property main_repo: RepoFor[Any]

Main repository injected by the factory.

abstractmethod async execute(*args, **kwargs)[source]

Execute core business logic.

Override with an explicit typed signature. The compiler inspects this method once at startup to build the ExecutionPlan.

Returns:

The result of the use case operation.

Parameters:
Return type:

ResultT