Skip to content

Validation Port

validation_port

Inbound port for domain command and query validation.

Responsibilities
  • Validate commands and queries against business rules.
  • Return structured validation errors.
Non-Responsibilities
  • Execute commands or queries.
  • Implement business logic directly.

ValidationPort

Bases: InboundPort

Inbound port for domain command and query validation.

Responsibilities
  • Inspect command and query payloads for rule violations.
  • Return a list of RuleViolationError instances.
Non-Responsibilities
  • Enforce authorization (handled by AuthorizationPort).
  • Modify command or query state.
Example 
class CreateOrderCommand:
    product_id: str
    quantity: int


class CreateOrderValidator(ValidationPort[CreateOrderCommand, object]):
    async def validate_command(
        self, command: Command[CreateOrderCommand]
    ) -> list[RuleViolationError]:
        return []

    ...
Source code in src/forging_blocks/application/ports/inbound/validation_port.py 
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 ValidationPort[CommandPayloadType, QueryPayloadType](InboundPort):
    """Inbound port for domain command and query validation.

    Responsibilities:
        - Inspect command and query payloads for rule violations.
        - Return a list of ``RuleViolationError`` instances.

    Non-Responsibilities:
        - Enforce authorization (handled by ``AuthorizationPort``).
        - Modify command or query state.

    Example:
        ```python
        class CreateOrderCommand:
            product_id: str
            quantity: int


        class CreateOrderValidator(ValidationPort[CreateOrderCommand, object]):
            async def validate_command(
                self, command: Command[CreateOrderCommand]
            ) -> list[RuleViolationError]:
                return []

            ...
        ```
    """

    @abstractmethod
    async def validate_command(
        self, command: Command[CommandPayloadType]
    ) -> list[RuleViolationError]:
        """Validate a domain command."""
        ...

    @abstractmethod
    async def validate_query(self, query: Query[QueryPayloadType]) -> list[RuleViolationError]:
        """Validate a domain query."""
        ...

validate_command(command: Command[CommandPayloadType]) -> list[RuleViolationError] abstractmethod async

Validate a domain command.

Source code in src/forging_blocks/application/ports/inbound/validation_port.py 
48
49
50
51
52
53
@abstractmethod
async def validate_command(
    self, command: Command[CommandPayloadType]
) -> list[RuleViolationError]:
    """Validate a domain command."""
    ...

validate_query(query: Query[QueryPayloadType]) -> list[RuleViolationError] abstractmethod async

Validate a domain query.

Source code in src/forging_blocks/application/ports/inbound/validation_port.py 
55
56
57
58
@abstractmethod
async def validate_query(self, query: Query[QueryPayloadType]) -> list[RuleViolationError]:
    """Validate a domain query."""
    ...