Skip to content

Application Service Port

application_service_port

Inbound application port for executing use cases and application services.

Use cases and application services define the application's orchestration
boundary: they coordinate domain operations to fulfill a specific
application intent. They are pure application-blocks logic and must not
depend on infrastructure details.

Responsibilities
  • Orchestrate domain interactions.
  • Call repositories, outbound ports, and aggregates.
  • Maintain transactional consistency (via Unit of Work).
Non-Responsibilities
  • Transport logic (HTTP, messaging systems).
  • Persistence implementation details.
  • UI or framework-specific concerns.

UseCasePort = ApplicationServicePort module-attribute

Type alias for ApplicationServicePort.

Both names refer to the same contract and are fully interchangeable.

ApplicationServicePort

Bases: InboundPort

Inbound port for defining application use case operations.

A ApplicationServicePort represents an application-level action that may
involve multiple domain objects and outbound interactions. Use cases
must remain free of infrastructure dependencies and must uphold
application-level invariants.

Example 
class CreateOrderUseCase(ApplicationServicePort[dict[str, object], Order]):
    async def execute(self, request: dict[str, object]) -> Order: ...
Source code in src/forging_blocks/application/ports/inbound/application_service_port.py 
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
class ApplicationServicePort[RequestType, ResponseType](
    InboundPort,
):
    """Inbound port for defining application use case operations.

    A ``ApplicationServicePort`` represents an application-level action that may
    involve multiple domain objects and outbound interactions. Use cases
    must remain free of infrastructure dependencies and must uphold
    application-level invariants.

    Example:
        ```python
        class CreateOrderUseCase(ApplicationServicePort[dict[str, object], Order]):
            async def execute(self, request: dict[str, object]) -> Order: ...
        ```
    """

    @abstractmethod
    async def execute(self, request: RequestType) -> ResponseType:
        """Execute the use case asynchronously.

        Args:
            request: The request DTO carrying user or system input.

        Returns:
            A DTO or domain object representing the outcome.

        Raises:
            ApplicationError: If the use case fails for domain reasons.

        Notes:
            This method is asynchronous and should not perform blocking
            operations.

        """
        ...

execute(request: RequestType) -> ResponseType abstractmethod async

Execute the use case asynchronously.

Parameters:

Name Type Description Default
request RequestType

The request DTO carrying user or system input.

 required

Returns:

Type Description
ResponseType

A DTO or domain object representing the outcome.

Raises:

Type Description
ApplicationError

If the use case fails for domain reasons.
Notes

This method is asynchronous and should not perform blocking
operations.

Source code in src/forging_blocks/application/ports/inbound/application_service_port.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@abstractmethod
async def execute(self, request: RequestType) -> ResponseType:
    """Execute the use case asynchronously.

    Args:
        request: The request DTO carrying user or system input.

    Returns:
        A DTO or domain object representing the outcome.

    Raises:
        ApplicationError: If the use case fails for domain reasons.

    Notes:
        This method is asynchronous and should not perform blocking
        operations.

    """
    ...