Skip to content

DTOs

The models/dtos subpackage contains Pydantic BaseModel data transfer objects used for API request/response shapes, pagination, sorting, search, and email payloads.

Base DTOs

Base DTO classes providing common fields and validators inherited by domain-specific DTOs.

Classes:

Name Description
BaseDTO

Base Data Transfer Object class.

BaseDTO

Base Data Transfer Object class.

This class extends Pydantic's BaseModel to provide common configuration for all DTOs in the application.

Source code in archipy/models/dtos/base_dtos.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class BaseDTO(BaseModel):
    """Base Data Transfer Object class.

    This class extends Pydantic's BaseModel to provide common configuration
    for all DTOs in the application.
    """

    model_config = ConfigDict(
        extra="ignore",
        validate_default=True,
        from_attributes=True,
        frozen=True,
        str_strip_whitespace=True,
        arbitrary_types_allowed=True,
    )

options: show_root_toc_entry: false heading_level: 3

Base Protobuf DTO

Base DTO class for Protobuf-backed data transfer objects, bridging gRPC and Pydantic models.

Classes:

Name Description
BaseProtobufDTO

A base DTO that can be converted to and from a Protobuf message.

BaseProtobufDTO

A base DTO that can be converted to and from a Protobuf message.

Requires 'google-protobuf' to be installed.

Methods:

Name Description
from_proto

Converts a Protobuf message into a Pydantic DTO instance.
to_proto

Converts the Pydantic DTO instance into a Protobuf message.
Source code in archipy/models/dtos/base_protobuf_dto.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
59
class BaseProtobufDTO(BaseDTO):
    """A base DTO that can be converted to and from a Protobuf message.

    Requires 'google-protobuf' to be installed.
    """

    _proto_class: ClassVar[type[Message] | None] = None

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        # Add a check at runtime when someone tries to use the class
        if not PROTOBUF_AVAILABLE:
            raise RuntimeError("The 'protobuf' extra is not installed. ")
        super().__init__(*args, **kwargs)

    @classmethod
    def from_proto(cls, request: Message) -> Self:
        """Converts a Protobuf message into a Pydantic DTO instance."""
        if cls._proto_class is None:
            raise NotImplementedError(f"{cls.__name__} is not mapped to a proto class.")

        if not isinstance(request, cls._proto_class):
            raise InvalidEntityTypeError(
                message=f"{cls.__name__}.from_proto expected a different type of request.",
                expected_type=cls._proto_class.__name__,
                actual_type=type(request).__name__,
            )

        input_data = MessageToDict(
            message=request,
            always_print_fields_with_no_presence=True,
            preserving_proto_field_name=True,
        )
        return cls.model_validate(input_data)

    def to_proto(self) -> Message:
        """Converts the Pydantic DTO instance into a Protobuf message."""
        if self._proto_class is None:
            raise NotImplementedError(f"{self.__class__.__name__} is not mapped to a proto class.")

        return ParseDict(self.model_dump(mode="json"), self._proto_class())

from_proto classmethod 

from_proto(request: Message) -> Self

Converts a Protobuf message into a Pydantic DTO instance.

Source code in archipy/models/dtos/base_protobuf_dto.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@classmethod
def from_proto(cls, request: Message) -> Self:
    """Converts a Protobuf message into a Pydantic DTO instance."""
    if cls._proto_class is None:
        raise NotImplementedError(f"{cls.__name__} is not mapped to a proto class.")

    if not isinstance(request, cls._proto_class):
        raise InvalidEntityTypeError(
            message=f"{cls.__name__}.from_proto expected a different type of request.",
            expected_type=cls._proto_class.__name__,
            actual_type=type(request).__name__,
        )

    input_data = MessageToDict(
        message=request,
        always_print_fields_with_no_presence=True,
        preserving_proto_field_name=True,
    )
    return cls.model_validate(input_data)

to_proto 

to_proto() -> Message

Converts the Pydantic DTO instance into a Protobuf message.

Source code in archipy/models/dtos/base_protobuf_dto.py 
54
55
56
57
58
59
def to_proto(self) -> Message:
    """Converts the Pydantic DTO instance into a Protobuf message."""
    if self._proto_class is None:
        raise NotImplementedError(f"{self.__class__.__name__} is not mapped to a proto class.")

    return ParseDict(self.model_dump(mode="json"), self._proto_class())

options: show_root_toc_entry: false heading_level: 3

Pagination DTO

DTOs for pagination input and output, including page number, page size, and total count fields.

Classes:

Name Description
PaginationDTO

Data Transfer Object for pagination parameters.

PaginationDTO

Data Transfer Object for pagination parameters.

This DTO encapsulates pagination information for database queries and API responses, providing a standard way to specify which subset of results to retrieve.

Attributes:

Name Type Description
page int

The current page number (1-based indexing)
page_size int

Number of items per page
offset int

Calculated offset for database queries based on page and page_size

Examples:

>>> from archipy.models.dtos.pagination_dto import PaginationDTO
>>>
>>> # Default pagination (page 1, 10 items per page)
>>> pagination = PaginationDTO()
>>>
>>> # Custom pagination
>>> pagination = PaginationDTO(page=2, page_size=25)
>>> print(pagination.offset)  # Access offset as a property
25
>>>
>>> # Using with a database query
>>> def get_users(pagination: PaginationDTO):
...     query = select(User).offset(pagination.offset).limit(pagination.page_size)
...     return db.execute(query).scalars().all()

Methods:

Name Description
validate_pagination

Validate pagination limits to prevent excessive resource usage.
Source code in archipy/models/dtos/pagination_dto.py 
13
14
15
16
17
18
19
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class PaginationDTO(BaseDTO):
    """Data Transfer Object for pagination parameters.

    This DTO encapsulates pagination information for database queries and API responses,
    providing a standard way to specify which subset of results to retrieve.

    Attributes:
        page (int): The current page number (1-based indexing)
        page_size (int): Number of items per page
        offset (int): Calculated offset for database queries based on page and page_size

    Examples:
        >>> from archipy.models.dtos.pagination_dto import PaginationDTO
        >>>
        >>> # Default pagination (page 1, 10 items per page)
        >>> pagination = PaginationDTO()
        >>>
        >>> # Custom pagination
        >>> pagination = PaginationDTO(page=2, page_size=25)
        >>> print(pagination.offset)  # Access offset as a property
        25
        >>>
        >>> # Using with a database query
        >>> def get_users(pagination: PaginationDTO):
        ...     query = select(User).offset(pagination.offset).limit(pagination.page_size)
        ...     return db.execute(query).scalars().all()
    """

    page: int = Field(default=1, ge=1, description="Page number (1-indexed)")
    page_size: int = Field(default=10, ge=1, le=100, description="Number of items per page")

    MAX_ITEMS: ClassVar = 10000

    @model_validator(mode="after")
    def validate_pagination(self) -> Self:
        """Validate pagination limits to prevent excessive resource usage.

        Ensures that the requested number of items (page * page_size) doesn't exceed
        the maximum allowed limit.

        Returns:
            The validated model instance if valid.

        Raises:
            OutOfRangeError: If the total requested items exceeds MAX_ITEMS.
        """
        total_items = self.page * self.page_size
        if total_items > self.MAX_ITEMS:
            raise OutOfRangeError(field_name="pagination")
        return self

    @property
    def offset(self) -> int:
        """Calculate the offset for database queries.

        This property calculates how many records to skip based on the
        current page and page size.

        Returns:
            int: The number of records to skip

        Examples:
            >>> pagination = PaginationDTO(page=3, page_size=20)
            >>> pagination.offset
            40  # Skip the first 40 records (2 pages of 20 items)
        """
        return (self.page - 1) * self.page_size

offset property 

offset: int

Calculate the offset for database queries.

This property calculates how many records to skip based on the current page and page size.

Returns:

Name Type Description
int int

The number of records to skip

Examples:

>>> pagination = PaginationDTO(page=3, page_size=20)
>>> pagination.offset
40  # Skip the first 40 records (2 pages of 20 items)

validate_pagination 

validate_pagination() -> Self

Validate pagination limits to prevent excessive resource usage.

Ensures that the requested number of items (page * page_size) doesn't exceed the maximum allowed limit.

Returns:

Type Description
Self

The validated model instance if valid.

Raises:

Type Description
OutOfRangeError

If the total requested items exceeds MAX_ITEMS.
Source code in archipy/models/dtos/pagination_dto.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@model_validator(mode="after")
def validate_pagination(self) -> Self:
    """Validate pagination limits to prevent excessive resource usage.

    Ensures that the requested number of items (page * page_size) doesn't exceed
    the maximum allowed limit.

    Returns:
        The validated model instance if valid.

    Raises:
        OutOfRangeError: If the total requested items exceeds MAX_ITEMS.
    """
    total_items = self.page * self.page_size
    if total_items > self.MAX_ITEMS:
        raise OutOfRangeError(field_name="pagination")
    return self

options: show_root_toc_entry: false heading_level: 3

Sort DTO

DTOs for expressing sort order in list/search requests.

Classes:

Name Description
SortDTO

Data Transfer Object for sorting parameters.

SortDTO

Data Transfer Object for sorting parameters.

This DTO encapsulates sorting information for database queries and API responses, providing a standard way to specify how results should be ordered.

Attributes:

Name Type Description
column T | str

The name or enum value of the column to sort by
order str

The sort direction - "ASC" for ascending, "DESC" for descending

Examples:

>>> from archipy.models.dtos.sort_dto import SortDTO
>>> from archipy.models.types.sort_order_type import SortOrderType
>>>
>>> # Sort by name in ascending order
>>> sort = SortDTO(column="name", order=SortOrderType.ASCENDING)
>>>
>>> # Sort by creation date in descending order (newest first)
>>> sort = SortDTO(column="created_at", order="DESCENDING")
>>>
>>> # Using with a database query
>>> def get_sorted_users(sort: SortDTO = SortDTO.default()):
...     query = select(User)
...     if sort.order == SortOrderType.ASCENDING:
...         query = query.order_by(getattr(User, sort.column).asc())
...     else:
...         query = query.order_by(getattr(User, sort.column).desc())
...     return db.execute(query).scalars().all()
>>>
>>> # Using with enum column types
>>> from enum import Enum
>>> class UserColumns(Enum):
...     ID = "id"
...     NAME = "name"
...     EMAIL = "email"
...     CREATED_AT = "created_at"
>>>
>>> # Create a sort configuration with enum
>>> sort = SortDTO[UserColumns](column=UserColumns.NAME, order=SortOrderType.ASCENDING)

Methods:

Name Description
default

Create a default sort configuration.
Source code in archipy/models/dtos/sort_dto.py 
12
13
14
15
16
17
18
19
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
59
60
61
62
63
64
65
66
67
68
69
70
71
class SortDTO[T](BaseModel):
    """Data Transfer Object for sorting parameters.

    This DTO encapsulates sorting information for database queries and API responses,
    providing a standard way to specify how results should be ordered.

    Attributes:
        column (T | str): The name or enum value of the column to sort by
        order (str): The sort direction - "ASC" for ascending, "DESC" for descending

    Examples:
        >>> from archipy.models.dtos.sort_dto import SortDTO
        >>> from archipy.models.types.sort_order_type import SortOrderType
        >>>
        >>> # Sort by name in ascending order
        >>> sort = SortDTO(column="name", order=SortOrderType.ASCENDING)
        >>>
        >>> # Sort by creation date in descending order (newest first)
        >>> sort = SortDTO(column="created_at", order="DESCENDING")
        >>>
        >>> # Using with a database query
        >>> def get_sorted_users(sort: SortDTO = SortDTO.default()):
        ...     query = select(User)
        ...     if sort.order == SortOrderType.ASCENDING:
        ...         query = query.order_by(getattr(User, sort.column).asc())
        ...     else:
        ...         query = query.order_by(getattr(User, sort.column).desc())
        ...     return db.execute(query).scalars().all()
        >>>
        >>> # Using with enum column types
        >>> from enum import Enum
        >>> class UserColumns(Enum):
        ...     ID = "id"
        ...     NAME = "name"
        ...     EMAIL = "email"
        ...     CREATED_AT = "created_at"
        >>>
        >>> # Create a sort configuration with enum
        >>> sort = SortDTO[UserColumns](column=UserColumns.NAME, order=SortOrderType.ASCENDING)
    """

    column: T | str = Field(default="created_at", description="Column name or enum to sort by")
    order: SortOrderType = Field(default=SortOrderType.DESCENDING, description="Sort order (ASCENDING or DESCENDING)")

    @classmethod
    def default(cls) -> SortDTO:
        """Create a default sort configuration.

        Returns a sort configuration that orders by created_at in descending order
        (newest first), which is a common default sorting behavior.

        Returns:
            SortDTO: A default sort configuration

        Examples:
            >>> default_sort = SortDTO.default()
            >>> print(f"Sort by {default_sort.column} {default_sort.order}")
            Sort by created_at DESCENDING
        """
        return cls(column="created_at", order=SortOrderType.DESCENDING)

default classmethod 

default() -> SortDTO

Create a default sort configuration.

Returns a sort configuration that orders by created_at in descending order (newest first), which is a common default sorting behavior.

Returns:

Name Type Description
SortDTO SortDTO

A default sort configuration

Examples:

>>> default_sort = SortDTO.default()
>>> print(f"Sort by {default_sort.column} {default_sort.order}")
Sort by created_at DESCENDING
Source code in archipy/models/dtos/sort_dto.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def default(cls) -> SortDTO:
    """Create a default sort configuration.

    Returns a sort configuration that orders by created_at in descending order
    (newest first), which is a common default sorting behavior.

    Returns:
        SortDTO: A default sort configuration

    Examples:
        >>> default_sort = SortDTO.default()
        >>> print(f"Sort by {default_sort.column} {default_sort.order}")
        Sort by created_at DESCENDING
    """
    return cls(column="created_at", order=SortOrderType.DESCENDING)

options: show_root_toc_entry: false heading_level: 3

Search Input DTO

DTO for structured search input combining filter, sort, and pagination parameters.

Classes:

Name Description
SearchInputDTO

Data Transfer Object for search inputs with pagination and sorting.

SearchInputDTO

Data Transfer Object for search inputs with pagination and sorting.

This DTO encapsulates search parameters for database queries and API responses, providing a standard way to handle pagination and sorting.

Class Type Parameters:

Name Bound or Constraints Description Default
T

The type for sort column (usually an Enum with column names).

 required
Source code in archipy/models/dtos/search_input_dto.py 
13
14
15
16
17
18
19
20
21
22
23
24
class SearchInputDTO[T](BaseModel):
    """Data Transfer Object for search inputs with pagination and sorting.

    This DTO encapsulates search parameters for database queries and API responses,
    providing a standard way to handle pagination and sorting.

    Type Parameters:
        T: The type for sort column (usually an Enum with column names).
    """

    pagination: PaginationDTO | None = None
    sort_info: SortDTO[T] | None = None

options: show_root_toc_entry: false heading_level: 3

Range DTOs

DTOs for expressing numeric and date range filters in queries.

Classes:

Name Description
Comparable

Protocol for types that support comparison operators.
BaseRangeDTO

Base Data Transfer Object for range queries.
DecimalRangeDTO

Data Transfer Object for decimal range queries.
IntegerRangeDTO

Data Transfer Object for integer range queries.
DateRangeDTO

Data Transfer Object for date range queries.
DatetimeRangeDTO

Data Transfer Object for datetime range queries.
DatetimeIntervalRangeDTO

Data Transfer Object for datetime range queries with interval.

Comparable

Protocol for types that support comparison operators.

Source code in archipy/models/dtos/range_dtos.py 
13
14
15
16
17
18
class Comparable(Protocol):
    """Protocol for types that support comparison operators."""

    def __gt__(self, other: object) -> bool:
        """Greater than comparison operator."""
        ...

BaseRangeDTO

Base Data Transfer Object for range queries.

Encapsulates a range of values with from_ and to fields. Provides validation to ensure range integrity.

Methods:

Name Description
validate_range

Validate that from_ is less than or equal to to when both are provided.
Source code in archipy/models/dtos/range_dtos.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
class BaseRangeDTO[R](BaseDTO):
    """Base Data Transfer Object for range queries.

    Encapsulates a range of values with from_ and to fields.
    Provides validation to ensure range integrity.
    """

    from_: R | None = None
    to: R | None = None

    @model_validator(mode="after")
    def validate_range(self) -> Self:
        """Validate that from_ is less than or equal to to when both are provided.

        Returns:
            Self: The validated model instance.

        Raises:
            OutOfRangeError: If from_ is greater than to.
        """
        if self.from_ is not None and self.to is not None:
            # Use comparison with proper type handling
            # The protocol ensures both values support comparison
            try:
                if self.from_ > self.to:  # ty: ignore[unsupported-operator]
                    raise OutOfRangeError(field_name="from_")
            except TypeError:
                # If comparison fails, skip validation (shouldn't happen with proper types)
                pass
        return self

validate_range 

validate_range() -> Self

Validate that from_ is less than or equal to to when both are provided.

Returns:

Name Type Description
Self Self

The validated model instance.

Raises:

Type Description
OutOfRangeError

If from_ is greater than to.
Source code in archipy/models/dtos/range_dtos.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@model_validator(mode="after")
def validate_range(self) -> Self:
    """Validate that from_ is less than or equal to to when both are provided.

    Returns:
        Self: The validated model instance.

    Raises:
        OutOfRangeError: If from_ is greater than to.
    """
    if self.from_ is not None and self.to is not None:
        # Use comparison with proper type handling
        # The protocol ensures both values support comparison
        try:
            if self.from_ > self.to:  # ty: ignore[unsupported-operator]
                raise OutOfRangeError(field_name="from_")
        except TypeError:
            # If comparison fails, skip validation (shouldn't happen with proper types)
            pass
    return self

DecimalRangeDTO

Data Transfer Object for decimal range queries.

Methods:

Name Description
convert_to_decimal

Convert input values to Decimal type.
Source code in archipy/models/dtos/range_dtos.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class DecimalRangeDTO(BaseRangeDTO[Decimal]):
    """Data Transfer Object for decimal range queries."""

    from_: Decimal | None = None
    to: Decimal | None = None

    @field_validator("from_", "to", mode="before")
    @classmethod
    def convert_to_decimal(cls, value: Decimal | str | None) -> Decimal | None:
        """Convert input values to Decimal type.

        Args:
            value: The value to convert (None, string, or Decimal).

        Returns:
            Decimal | None: The converted Decimal value or None.

        Raises:
            InvalidArgumentError: If the value cannot be converted to Decimal.
        """
        if value is None:
            return None
        try:
            return Decimal(value)
        except (TypeError, ValueError) as e:
            raise InvalidArgumentError(argument_name="value") from e

convert_to_decimal classmethod 

convert_to_decimal(
    value: Decimal | str | None,
) -> Decimal | None

Convert input values to Decimal type.

Parameters:

Name Type Description Default
value Decimal | str | None

The value to convert (None, string, or Decimal).

 required

Returns:

Type Description
Decimal | None

Decimal | None: The converted Decimal value or None.

Raises:

Type Description
InvalidArgumentError

If the value cannot be converted to Decimal.
Source code in archipy/models/dtos/range_dtos.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@field_validator("from_", "to", mode="before")
@classmethod
def convert_to_decimal(cls, value: Decimal | str | None) -> Decimal | None:
    """Convert input values to Decimal type.

    Args:
        value: The value to convert (None, string, or Decimal).

    Returns:
        Decimal | None: The converted Decimal value or None.

    Raises:
        InvalidArgumentError: If the value cannot be converted to Decimal.
    """
    if value is None:
        return None
    try:
        return Decimal(value)
    except (TypeError, ValueError) as e:
        raise InvalidArgumentError(argument_name="value") from e

IntegerRangeDTO

Data Transfer Object for integer range queries.

Source code in archipy/models/dtos/range_dtos.py 
84
85
86
87
88
class IntegerRangeDTO(BaseRangeDTO[int]):
    """Data Transfer Object for integer range queries."""

    from_: int | None = None
    to: int | None = None

DateRangeDTO

Data Transfer Object for date range queries.

Source code in archipy/models/dtos/range_dtos.py 
91
92
93
94
95
class DateRangeDTO(BaseRangeDTO[date]):
    """Data Transfer Object for date range queries."""

    from_: date | None = None
    to: date | None = None

DatetimeRangeDTO

Data Transfer Object for datetime range queries.

Source code in archipy/models/dtos/range_dtos.py 
 98
 99
100
101
102
class DatetimeRangeDTO(BaseRangeDTO[datetime]):
    """Data Transfer Object for datetime range queries."""

    from_: datetime | None = None
    to: datetime | None = None

DatetimeIntervalRangeDTO

Data Transfer Object for datetime range queries with interval.

Rejects requests if the number of intervals exceeds MAX_ITEMS or if interval-specific range size or 'to' age constraints are violated.

Methods:

Name Description
validate_interval_constraints

Validate interval based on range size, 'to' field age, and max intervals.
Source code in archipy/models/dtos/range_dtos.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
class DatetimeIntervalRangeDTO(BaseRangeDTO[datetime]):
    """Data Transfer Object for datetime range queries with interval.

    Rejects requests if the number of intervals exceeds MAX_ITEMS or if interval-specific
    range size or 'to' age constraints are violated.
    """

    from_: datetime
    to: datetime
    interval: TimeIntervalUnitType

    # Maximum number of intervals allowed
    MAX_ITEMS: ClassVar[int] = 100

    # Range size limits for each interval
    RANGE_SIZE_LIMITS: ClassVar[dict[TimeIntervalUnitType, timedelta]] = {
        TimeIntervalUnitType.SECONDS: timedelta(days=2),
        TimeIntervalUnitType.MINUTES: timedelta(days=7),
        TimeIntervalUnitType.HOURS: timedelta(days=30),
        TimeIntervalUnitType.DAYS: timedelta(days=365),
        TimeIntervalUnitType.WEEKS: timedelta(days=365 * 2),
        TimeIntervalUnitType.MONTHS: timedelta(days=365 * 5),  # No limit for MONTHS, set high
        TimeIntervalUnitType.YEAR: timedelta(days=365 * 10),  # No limit for YEAR, set high
    }

    # 'to' age limits for each interval
    TO_AGE_LIMITS: ClassVar[dict[TimeIntervalUnitType, timedelta]] = {
        TimeIntervalUnitType.SECONDS: timedelta(days=2),
        TimeIntervalUnitType.MINUTES: timedelta(days=7),
        TimeIntervalUnitType.HOURS: timedelta(days=30),
        TimeIntervalUnitType.DAYS: timedelta(days=365 * 5),
        TimeIntervalUnitType.WEEKS: timedelta(days=365 * 10),
        TimeIntervalUnitType.MONTHS: timedelta(days=365 * 20),  # No limit for MONTHS, set high
        TimeIntervalUnitType.YEAR: timedelta(days=365 * 50),  # No limit for YEAR, set high
    }

    # Mapping of intervals to timedelta for step size
    INTERVAL_TO_TIMEDELTA: ClassVar[dict[TimeIntervalUnitType, timedelta]] = {
        TimeIntervalUnitType.SECONDS: timedelta(seconds=1),
        TimeIntervalUnitType.MINUTES: timedelta(minutes=1),
        TimeIntervalUnitType.HOURS: timedelta(hours=1),
        TimeIntervalUnitType.DAYS: timedelta(days=1),
        TimeIntervalUnitType.WEEKS: timedelta(weeks=1),
        TimeIntervalUnitType.MONTHS: timedelta(days=30),  # Approximate
        TimeIntervalUnitType.YEAR: timedelta(days=365),  # Approximate
    }

    @model_validator(mode="after")
    def validate_interval_constraints(self) -> Self:
        """Validate interval based on range size, 'to' field age, and max intervals.

        - Each interval has specific range size and 'to' age limits.
        - Rejects if the number of intervals exceeds MAX_ITEMS.

        Returns:
            Self: The validated model instance.

        Raises:
            OutOfRangeError: If interval constraints are violated or number of intervals > MAX_ITEMS.
        """
        if self.from_ is not None and self.to is not None:
            # Validate range size limit for the selected interval
            range_size = self.to - self.from_
            max_range_size = self.RANGE_SIZE_LIMITS.get(self.interval)
            if max_range_size and range_size > max_range_size:
                raise OutOfRangeError(field_name="range_size")

            # Validate 'to' age limit
            current_time = datetime.now()
            max_to_age = self.TO_AGE_LIMITS.get(self.interval)
            if max_to_age:
                age_threshold = current_time - max_to_age
                if self.to < age_threshold:
                    raise OutOfRangeError(field_name="to")

            # Calculate number of intervals
            step = self.INTERVAL_TO_TIMEDELTA[self.interval]
            range_duration = self.to - self.from_
            num_intervals = int(range_duration.total_seconds() / step.total_seconds()) + 1

            # Reject if number of intervals exceeds MAX_ITEMS
            if num_intervals > self.MAX_ITEMS:
                raise OutOfRangeError(field_name="interval_count")

        return self

validate_interval_constraints 

validate_interval_constraints() -> Self

Validate interval based on range size, 'to' field age, and max intervals.

  • Each interval has specific range size and 'to' age limits.
  • Rejects if the number of intervals exceeds MAX_ITEMS.

Returns:

Name Type Description
Self Self

The validated model instance.

Raises:

Type Description
OutOfRangeError

If interval constraints are violated or number of intervals > MAX_ITEMS.
Source code in archipy/models/dtos/range_dtos.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
@model_validator(mode="after")
def validate_interval_constraints(self) -> Self:
    """Validate interval based on range size, 'to' field age, and max intervals.

    - Each interval has specific range size and 'to' age limits.
    - Rejects if the number of intervals exceeds MAX_ITEMS.

    Returns:
        Self: The validated model instance.

    Raises:
        OutOfRangeError: If interval constraints are violated or number of intervals > MAX_ITEMS.
    """
    if self.from_ is not None and self.to is not None:
        # Validate range size limit for the selected interval
        range_size = self.to - self.from_
        max_range_size = self.RANGE_SIZE_LIMITS.get(self.interval)
        if max_range_size and range_size > max_range_size:
            raise OutOfRangeError(field_name="range_size")

        # Validate 'to' age limit
        current_time = datetime.now()
        max_to_age = self.TO_AGE_LIMITS.get(self.interval)
        if max_to_age:
            age_threshold = current_time - max_to_age
            if self.to < age_threshold:
                raise OutOfRangeError(field_name="to")

        # Calculate number of intervals
        step = self.INTERVAL_TO_TIMEDELTA[self.interval]
        range_duration = self.to - self.from_
        num_intervals = int(range_duration.total_seconds() / step.total_seconds()) + 1

        # Reject if number of intervals exceeds MAX_ITEMS
        if num_intervals > self.MAX_ITEMS:
            raise OutOfRangeError(field_name="interval_count")

    return self

options: show_root_toc_entry: false heading_level: 3

Email DTOs

DTOs for composing email messages including recipients, subject, body, and attachments.

Classes:

Name Description
EmailAttachmentDTO

Pydantic model for email attachments.

EmailAttachmentDTO

Pydantic model for email attachments.

Methods:

Name Description
validate_attachment

Validate and normalize attachment fields.
Source code in archipy/models/dtos/email_dtos.py 
10
11
12
13
14
15
16
17
18
19
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
class EmailAttachmentDTO(BaseDTO):
    """Pydantic model for email attachments."""

    content: str | bytes | BinaryIO
    filename: str
    content_type: str | None = Field(default=None)
    content_disposition: EmailAttachmentDispositionType = Field(default=EmailAttachmentDispositionType.ATTACHMENT)
    content_id: str | None = Field(default=None)
    attachment_type: EmailAttachmentType
    max_size: int

    @model_validator(mode="after")
    def validate_attachment(self) -> Self:
        """Validate and normalize attachment fields.

        This validator performs three operations:
        1. Sets content_type based on filename extension if not provided
        2. Validates that attachment size does not exceed maximum allowed size
        3. Ensures content_id is properly formatted with angle brackets

        Returns:
            The validated model instance

        Raises:
            ValueError: If attachment size exceeds maximum allowed size
        """
        # Set content type from filename if not provided
        if self.content_type is None:
            content_type, _ = mimetypes.guess_type(self.filename)
            self.content_type = content_type or "application/octet-stream"

        # Validate attachment size
        content = self.content
        if isinstance(content, str | bytes):
            content_size = len(content)
            if content_size > self.max_size:
                error_msg = f"Attachment size exceeds maximum allowed size of {self.max_size} bytes"
                raise ValueError(error_msg)

        # Ensure content_id has angle brackets
        if self.content_id and not self.content_id.startswith("<"):
            self.content_id = f"<{self.content_id}>"

        return self

validate_attachment 

validate_attachment() -> Self

Validate and normalize attachment fields.

This validator performs three operations: 1. Sets content_type based on filename extension if not provided 2. Validates that attachment size does not exceed maximum allowed size 3. Ensures content_id is properly formatted with angle brackets

Returns:

Type Description
Self

The validated model instance

Raises:

Type Description
ValueError

If attachment size exceeds maximum allowed size
Source code in archipy/models/dtos/email_dtos.py 
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
@model_validator(mode="after")
def validate_attachment(self) -> Self:
    """Validate and normalize attachment fields.

    This validator performs three operations:
    1. Sets content_type based on filename extension if not provided
    2. Validates that attachment size does not exceed maximum allowed size
    3. Ensures content_id is properly formatted with angle brackets

    Returns:
        The validated model instance

    Raises:
        ValueError: If attachment size exceeds maximum allowed size
    """
    # Set content type from filename if not provided
    if self.content_type is None:
        content_type, _ = mimetypes.guess_type(self.filename)
        self.content_type = content_type or "application/octet-stream"

    # Validate attachment size
    content = self.content
    if isinstance(content, str | bytes):
        content_size = len(content)
        if content_size > self.max_size:
            error_msg = f"Attachment size exceeds maximum allowed size of {self.max_size} bytes"
            raise ValueError(error_msg)

    # Ensure content_id has angle brackets
    if self.content_id and not self.content_id.startswith("<"):
        self.content_id = f"<{self.content_id}>"

    return self

options: show_root_toc_entry: false heading_level: 3

FastAPI Exception Response DTO

DTO representing the standardized error response body returned by FastAPI exception handlers.

Classes:

Name Description
FastAPIErrorResponseDTO

Standardized error response model for OpenAPI documentation.
ValidationErrorResponseDTO

Specific response model for validation errors.

FastAPIErrorResponseDTO

Standardized error response model for OpenAPI documentation.

Source code in archipy/models/dtos/fastapi_exception_response_dto.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
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
59
60
61
class FastAPIErrorResponseDTO:
    """Standardized error response model for OpenAPI documentation."""

    def __init__(self, exception: type[BaseError], additional_properties: dict | None = None) -> None:
        """Initialize the error response model.

        Args:
            exception: The exception class (not instance) with error details as class attributes
            additional_properties: Additional properties to include in the response
        """
        self.status_code = exception.http_status

        # Base properties that all errors have
        detail_properties = {
            "code": {"type": "string", "example": exception.code, "description": "Error code identifier"},
            "message_en": {
                "type": "string",
                "example": exception.message_en,
                "description": "Error message in English",
            },
            "message_fa": {
                "type": "string",
                "example": exception.message_fa,
                "description": "Error message in Persian",
            },
            "http_status": {"type": "integer", "example": exception.http_status, "description": "HTTP status code"},
        }

        # Add additional properties if provided
        if additional_properties:
            detail_properties.update(additional_properties)

        self.model = {
            "description": exception.message_en,
            "content": {
                "application/json": {
                    "schema": {
                        "type": "object",
                        "properties": {
                            "error": {
                                "type": "string",
                                "example": exception.code,
                                "description": "Error code identifier",
                            },
                            "detail": {
                                "type": "object",
                                "properties": detail_properties,
                                "required": ["code", "message_en", "message_fa", "http_status"],
                                "additionalProperties": False,
                                "description": "Detailed error information",
                            },
                        },
                    },
                },
            },
        }

ValidationErrorResponseDTO

Specific response model for validation errors.

Source code in archipy/models/dtos/fastapi_exception_response_dto.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class ValidationErrorResponseDTO(FastAPIErrorResponseDTO):
    """Specific response model for validation errors."""

    def __init__(self) -> None:
        """Initialize the validation error response model."""
        self.status_code = HTTPStatus.UNPROCESSABLE_ENTITY
        self.model = {
            "description": "Validation Error",
            "content": {
                "application/json": {
                    "schema": {
                        "type": "object",
                        "properties": {
                            "error": {
                                "type": "string",
                                "example": "VALIDATION_ERROR",
                                "description": "Error code identifier",
                            },
                            "detail": {
                                "type": "array",
                                "items": {
                                    "type": "object",
                                    "properties": {
                                        "field": {
                                            "type": "string",
                                            "example": "email",
                                            "description": "Field name that failed validation",
                                        },
                                        "message": {
                                            "type": "string",
                                            "example": "Invalid email format",
                                            "description": "Validation error message",
                                        },
                                        "value": {
                                            "type": "string",
                                            "example": "invalid@email",
                                            "description": "Invalid value that caused the error",
                                        },
                                    },
                                },
                                "example": [
                                    {"field": "email", "message": "Invalid email format", "value": "invalid@email"},
                                    {
                                        "field": "password",
                                        "message": "Password must be at least 8 characters",
                                        "value": "123",
                                    },
                                ],
                            },
                        },
                    },
                },
            },
        }

options: show_root_toc_entry: false heading_level: 3