Skip to content

castlecraft_engineer.exc

castlecraft_engineer.exc

AggregateNotFoundError

Bases: RepositoryError

Aggregate not found.

Source code in src/castlecraft_engineer/abstractions/repository.py 
27
28
29
30
31
32
class AggregateNotFoundError(RepositoryError):
    """Aggregate not found."""

    def __init__(self, aggregate_id: Any):
        super().__init__(f"Aggregate ID '{aggregate_id}' not found.")
        self.aggregate_id = aggregate_id

AuthorizationError

Bases: Exception

Custom exception for authorization failures.

Source code in src/castlecraft_engineer/authorization/base_service.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
class AuthorizationError(Exception):
    """Custom exception for authorization failures."""

    def __init__(
        self,
        subject_id: Optional[str],
        required_permissions: List[Permission],
        message: str = "Forbidden",
    ):
        self.subject_id = subject_id
        self.required_permissions = required_permissions
        super().__init__(
            f"Subject '{subject_id}' is not authorized for required "
            f"permissions: {required_permissions}. Reason: {message}"
        )

CommandHandlerNotFoundError

Bases: Exception

Raised when no handler is found for a given command type.

Source code in src/castlecraft_engineer/abstractions/command_bus.py 
15
16
17
18
19
20
21
22
class CommandHandlerNotFoundError(Exception):
    """Raised when no handler is found for a given command type."""

    def __init__(self, command_type: Type[Command]):
        super().__init__(
            f"No handler registered for command type {command_type.__name__}"
        )
        self.command_type = command_type

EventStoreConflictError

Bases: RuntimeError

Raised when there's a conflict appending events, e.g., due to version mismatch.

Source code in src/castlecraft_engineer/abstractions/event_store.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class EventStoreConflictError(RuntimeError):
    """Raised when there's a conflict appending events, e.g., due to version mismatch."""

    def __init__(
        self,
        aggregate_id: TAggregateId,
        expected_version: int,
        actual_version: Optional[int] = None,
    ):
        self.aggregate_id = aggregate_id
        self.expected_version = expected_version
        self.actual_version = actual_version
        super().__init__(
            f"Conflict appending events for aggregate {aggregate_id}. "
            f"Expected version {expected_version}, but current version is {actual_version}."
        )

HTTPError

Bases: IOError

Custom exception for HTTP errors, optionally holding the response.

Source code in src/castlecraft_engineer/common/requests.py 
91
92
93
94
95
96
97
98
class HTTPError(IOError):
    """
    Custom exception for HTTP errors, optionally holding the response.
    """

    def __init__(self, *args, response: Optional[Response] = None, **kwargs):
        super().__init__(*args, **kwargs)
        self.response = response

InvalidEncryptionFormat

Bases: ValueError

Custom exception for invalid encryption format.

Source code in src/castlecraft_engineer/common/crypto.py 
15
16
17
18
19
20
21
22
23
24
25
class InvalidEncryptionFormat(ValueError):
    """Custom exception for invalid encryption format."""

    def __init__(
        self, message: str = "Invalid encrypted data format", error: str | None = None
    ):
        super().__init__(message)
        self.error = error or message

    def __str__(self) -> str:
        return str(self.error)

OptimisticConcurrencyError

Bases: RepositoryError, StaleDataError

Optimistic concurrency conflict.

Source code in src/castlecraft_engineer/abstractions/repository.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class OptimisticConcurrencyError(RepositoryError, StaleDataError):
    """Optimistic concurrency conflict."""

    def __init__(
        self,
        aggregate_id: Any,
        expected_version: int,
        actual_version: Optional[int] = None,
    ):
        msg = (
            f"Concurrency error for ID '{aggregate_id}'. "
            f"Expected version {expected_version}, "
            f"but found version {actual_version} in database."
        )
        super().__init__(msg)
        self.aggregate_id = aggregate_id
        self.expected_version = expected_version
        self.actual_version = actual_version

QueryHandlerNotFoundError

Bases: Exception

Raised when no handler is found for a given query type.

Source code in src/castlecraft_engineer/abstractions/query_bus.py 
15
16
17
18
19
20
21
22
class QueryHandlerNotFoundError(Exception):
    """Raised when no handler is found for a given query type."""

    def __init__(self, query_type: Type[Query]):
        super().__init__(
            f"No handler registered for query type {query_type.__name__}",
        )
        self.query_type = query_type

RepositoryError

Bases: Exception

Base repository error.

Source code in src/castlecraft_engineer/abstractions/repository.py 
23
24
class RepositoryError(Exception):
    """Base repository error."""