castlecraft_engineer.abstractions.command_handler

castlecraft_engineer.abstractions.command_handler

CommandHandler

Bases: Generic[TCommand], ABC

Abstract base class for command handlers. Each handler is responsible for processing a specific type of command.

Source code in src/castlecraft_engineer/abstractions/command_handler.py

class CommandHandler(Generic[TCommand], abc.ABC):
    """
    Abstract base class for command handlers.
    Each handler is responsible for processing
    a specific type of command.
    """

    async def authorize(
        self,
        subject_id: Optional[str] = None,
        permissions: List[Permission] = [],
        *args: Any,
        **kwargs: Any,
    ) -> Optional[bool]:
        """
        Optional pre-execution authorization check for the command.

        NOTE: This method is NOT automatically called by the default CommandBus.
        It serves as a convention or hook for developers to implement custom
        pre-authorization logic if they choose to call it explicitly from within
        their `execute` method, or if they are using a custom bus implementation
        that invokes it.

        The recommended pattern for authorization with the default bus is to use
        an injected `AuthorizationService` within the `execute` method.

        Args:
            subject_id: The ID of the subject.
            permissions: A list of Permission objects representing the
                         permissions granted to the subject.
            *args: Additional positional arguments.
            **kwargs: Additional keyword arguments for context.
        Returns:
            True if the subject has permission, False otherwise.
        """
        pass  # pragma: no cover

    @abc.abstractmethod
    async def execute(
        self,
        command: TCommand,
        *args,
        subject_id: Optional[str] = None,
        permissions: List[Permission] = [],
        **kwargs,
    ) -> Any:
        """
        Handles the execution logic for the given command.

        Args:
            command: The command instance to be processed.
            subject_id: The ID of the subject attempting
                        to execute the command. Optional.
            permissions: The permissions associated with the subject.

        Raises:
            NotImplementedError: This method must be implemented by concrete subclasses.
        """
        raise NotImplementedError

authorize(subject_id=None, permissions=[], *args, **kwargs) async

Optional pre-execution authorization check for the command.

NOTE: This method is NOT automatically called by the default CommandBus. It serves as a convention or hook for developers to implement custom pre-authorization logic if they choose to call it explicitly from within their execute method, or if they are using a custom bus implementation that invokes it.

The recommended pattern for authorization with the default bus is to use an injected AuthorizationService within the execute method.

Parameters:

Name	Type	Description	Default
subject_id	Optional[str]	The ID of the subject.	None
permissions	List[Permission]	A list of Permission objects representing the permissions granted to the subject.	[]
*args	Any	Additional positional arguments.	()
**kwargs	Any	Additional keyword arguments for context.	{}

Returns: True if the subject has permission, False otherwise.

Source code in src/castlecraft_engineer/abstractions/command_handler.py

async def authorize(
    self,
    subject_id: Optional[str] = None,
    permissions: List[Permission] = [],
    *args: Any,
    **kwargs: Any,
) -> Optional[bool]:
    """
    Optional pre-execution authorization check for the command.

    NOTE: This method is NOT automatically called by the default CommandBus.
    It serves as a convention or hook for developers to implement custom
    pre-authorization logic if they choose to call it explicitly from within
    their `execute` method, or if they are using a custom bus implementation
    that invokes it.

    The recommended pattern for authorization with the default bus is to use
    an injected `AuthorizationService` within the `execute` method.

    Args:
        subject_id: The ID of the subject.
        permissions: A list of Permission objects representing the
                     permissions granted to the subject.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments for context.
    Returns:
        True if the subject has permission, False otherwise.
    """
    pass  # pragma: no cover

execute(command, *args, subject_id=None, permissions=[], **kwargs) abstractmethod async

Handles the execution logic for the given command.

Parameters:

Name	Type	Description	Default
command	TCommand	The command instance to be processed.	required
subject_id	Optional[str]	The ID of the subject attempting to execute the command. Optional.	None
permissions	List[Permission]	The permissions associated with the subject.	[]

Raises:

Type	Description
NotImplementedError	This method must be implemented by concrete subclasses.

Source code in src/castlecraft_engineer/abstractions/command_handler.py

@abc.abstractmethod
async def execute(
    self,
    command: TCommand,
    *args,
    subject_id: Optional[str] = None,
    permissions: List[Permission] = [],
    **kwargs,
) -> Any:
    """
    Handles the execution logic for the given command.

    Args:
        command: The command instance to be processed.
        subject_id: The ID of the subject attempting
                    to execute the command. Optional.
        permissions: The permissions associated with the subject.

    Raises:
        NotImplementedError: This method must be implemented by concrete subclasses.
    """
    raise NotImplementedError