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.

archipy.models.dtos.base_dtos.T module-attribute 

T = TypeVar('T', bound=Enum)

archipy.models.dtos.base_dtos.BaseDTO

Bases: BaseModel

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,
    )

archipy.models.dtos.base_dtos.BaseDTO.model_config class-attribute instance-attribute 

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.

archipy.models.dtos.base_protobuf_dto.PROTOBUF_AVAILABLE module-attribute 

PROTOBUF_AVAILABLE = True

archipy.models.dtos.base_protobuf_dto.BaseProtobufDTO

Bases: BaseDTO

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

Requires 'google-protobuf' to be installed.

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())

archipy.models.dtos.base_protobuf_dto.BaseProtobufDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.base_protobuf_dto.BaseProtobufDTO.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)

archipy.models.dtos.base_protobuf_dto.BaseProtobufDTO.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.

archipy.models.dtos.pagination_dto.T module-attribute 

T = TypeVar('T', bound=Enum)

archipy.models.dtos.pagination_dto.PaginationDTO

Bases: 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:

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()
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

archipy.models.dtos.pagination_dto.PaginationDTO.page class-attribute instance-attribute 

page: int = Field(
    default=1, ge=1, description="Page number (1-indexed)"
)

archipy.models.dtos.pagination_dto.PaginationDTO.page_size class-attribute instance-attribute 

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

archipy.models.dtos.pagination_dto.PaginationDTO.MAX_ITEMS class-attribute instance-attribute 

MAX_ITEMS: ClassVar = 10000

archipy.models.dtos.pagination_dto.PaginationDTO.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)

archipy.models.dtos.pagination_dto.PaginationDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.pagination_dto.PaginationDTO.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.

archipy.models.dtos.sort_dto.T module-attribute 

T = TypeVar('T', bound=Enum)

archipy.models.dtos.sort_dto.SortDTO

Bases: 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:

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)
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)

archipy.models.dtos.sort_dto.SortDTO.column class-attribute instance-attribute 

column: T | str = Field(
    default="created_at",
    description="Column name or enum to sort by",
)

archipy.models.dtos.sort_dto.SortDTO.order class-attribute instance-attribute 

order: SortOrderType = Field(
    default=SortOrderType.DESCENDING,
    description="Sort order (ASCENDING or DESCENDING)",
)

archipy.models.dtos.sort_dto.SortDTO.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.

archipy.models.dtos.search_input_dto.T module-attribute 

T = TypeVar('T', bound=Enum)

archipy.models.dtos.search_input_dto.SearchInputDTO

Bases: 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.

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

archipy.models.dtos.search_input_dto.SearchInputDTO.pagination class-attribute instance-attribute 

pagination: PaginationDTO | None = None

archipy.models.dtos.search_input_dto.SearchInputDTO.sort_info class-attribute instance-attribute 

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.

archipy.models.dtos.range_dtos.R module-attribute 

R = TypeVar('R', bound=Comparable)

archipy.models.dtos.range_dtos.Comparable

Bases: Protocol

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."""
        ...

archipy.models.dtos.range_dtos.BaseRangeDTO

Bases: BaseDTO

Base Data Transfer Object for range queries.

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

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

archipy.models.dtos.range_dtos.BaseRangeDTO.from_ class-attribute instance-attribute 

from_: R | None = None

archipy.models.dtos.range_dtos.BaseRangeDTO.to class-attribute instance-attribute 

to: R | None = None

archipy.models.dtos.range_dtos.BaseRangeDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.range_dtos.BaseRangeDTO.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

archipy.models.dtos.range_dtos.DecimalRangeDTO

Bases: BaseRangeDTO[Decimal]

Data Transfer Object for decimal range queries.

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

archipy.models.dtos.range_dtos.DecimalRangeDTO.from_ class-attribute instance-attribute 

from_: Decimal | None = None

archipy.models.dtos.range_dtos.DecimalRangeDTO.to class-attribute instance-attribute 

to: Decimal | None = None

archipy.models.dtos.range_dtos.DecimalRangeDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.range_dtos.DecimalRangeDTO.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

archipy.models.dtos.range_dtos.DecimalRangeDTO.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

archipy.models.dtos.range_dtos.IntegerRangeDTO

Bases: BaseRangeDTO[int]

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

archipy.models.dtos.range_dtos.IntegerRangeDTO.from_ class-attribute instance-attribute 

from_: int | None = None

archipy.models.dtos.range_dtos.IntegerRangeDTO.to class-attribute instance-attribute 

to: int | None = None

archipy.models.dtos.range_dtos.IntegerRangeDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.range_dtos.IntegerRangeDTO.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

archipy.models.dtos.range_dtos.DateRangeDTO

Bases: BaseRangeDTO[date]

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

archipy.models.dtos.range_dtos.DateRangeDTO.from_ class-attribute instance-attribute 

from_: date | None = None

archipy.models.dtos.range_dtos.DateRangeDTO.to class-attribute instance-attribute 

to: date | None = None

archipy.models.dtos.range_dtos.DateRangeDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.range_dtos.DateRangeDTO.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

archipy.models.dtos.range_dtos.DatetimeRangeDTO

Bases: BaseRangeDTO[datetime]

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

archipy.models.dtos.range_dtos.DatetimeRangeDTO.from_ class-attribute instance-attribute 

from_: datetime | None = None

archipy.models.dtos.range_dtos.DatetimeRangeDTO.to class-attribute instance-attribute 

to: datetime | None = None

archipy.models.dtos.range_dtos.DatetimeRangeDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.range_dtos.DatetimeRangeDTO.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

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO

Bases: 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.

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

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO.from_ instance-attribute 

from_: datetime

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO.to instance-attribute 

to: datetime

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO.interval instance-attribute 

interval: TimeIntervalUnitType

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO.MAX_ITEMS class-attribute 

MAX_ITEMS: int = 100

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO.RANGE_SIZE_LIMITS class-attribute 

RANGE_SIZE_LIMITS: 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),
    TimeIntervalUnitType.YEAR: timedelta(days=365 * 10),
}

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO.TO_AGE_LIMITS class-attribute 

TO_AGE_LIMITS: 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),
    TimeIntervalUnitType.YEAR: timedelta(days=365 * 50),
}

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO.INTERVAL_TO_TIMEDELTA class-attribute 

INTERVAL_TO_TIMEDELTA: 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),
    TimeIntervalUnitType.YEAR: timedelta(days=365),
}

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO.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

archipy.models.dtos.range_dtos.DatetimeIntervalRangeDTO.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

options: show_root_toc_entry: false heading_level: 3

Email DTOs

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

archipy.models.dtos.email_dtos.EmailAttachmentDTO

Bases: BaseDTO

Pydantic model for email attachments.

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)
            object.__setattr__(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("<"):
            object.__setattr__(self, "content_id", f"<{self.content_id}>")

        return self

archipy.models.dtos.email_dtos.EmailAttachmentDTO.content instance-attribute 

content: str | bytes | BinaryIO

archipy.models.dtos.email_dtos.EmailAttachmentDTO.filename instance-attribute 

filename: str

archipy.models.dtos.email_dtos.EmailAttachmentDTO.content_type class-attribute instance-attribute 

content_type: str | None = Field(default=None)

archipy.models.dtos.email_dtos.EmailAttachmentDTO.content_disposition class-attribute instance-attribute 

content_disposition: EmailAttachmentDispositionType = Field(
    default=EmailAttachmentDispositionType.ATTACHMENT
)

archipy.models.dtos.email_dtos.EmailAttachmentDTO.content_id class-attribute instance-attribute 

content_id: str | None = Field(default=None)

archipy.models.dtos.email_dtos.EmailAttachmentDTO.attachment_type instance-attribute 

attachment_type: EmailAttachmentType

archipy.models.dtos.email_dtos.EmailAttachmentDTO.max_size instance-attribute 

max_size: int

archipy.models.dtos.email_dtos.EmailAttachmentDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.email_dtos.EmailAttachmentDTO.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)
        object.__setattr__(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("<"):
        object.__setattr__(self, "content_id", f"<{self.content_id}>")

    return self

options: show_root_toc_entry: false heading_level: 3

Rate Limit Window DTO

DTO for a single rate-limit tier: maximum calls within a time window.

Data transfer objects for rate-limit window tiers.

archipy.models.dtos.rate_limit_window_dto.RateLimitWindowDTO

Bases: BaseDTO

Data transfer object for a single rate-limit tier.

Attributes:

Name Type Description
calls_count int

Maximum allowed requests within the window.
window_ms int

Window duration in milliseconds.
Source code in archipy/models/dtos/rate_limit_window_dto.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class RateLimitWindowDTO(BaseDTO):
    """Data transfer object for a single rate-limit tier.

    Attributes:
        calls_count: Maximum allowed requests within the window.
        window_ms: Window duration in milliseconds.
    """

    calls_count: int = Field(ge=1, description="Maximum allowed requests within the window.")
    window_ms: int = Field(gt=0, description="Window duration in milliseconds.")

    @property
    def key_suffix(self) -> str:
        """Return a stable Redis key segment for this window tier."""
        return f"{self.calls_count}x{self.window_ms}ms"

archipy.models.dtos.rate_limit_window_dto.RateLimitWindowDTO.calls_count class-attribute instance-attribute 

calls_count: int = Field(
    ge=1,
    description="Maximum allowed requests within the window.",
)

archipy.models.dtos.rate_limit_window_dto.RateLimitWindowDTO.window_ms class-attribute instance-attribute 

window_ms: int = Field(
    gt=0, description="Window duration in milliseconds."
)

archipy.models.dtos.rate_limit_window_dto.RateLimitWindowDTO.key_suffix property 

key_suffix: str

Return a stable Redis key segment for this window tier.

archipy.models.dtos.rate_limit_window_dto.RateLimitWindowDTO.model_config class-attribute instance-attribute 

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

FastAPI Exception Response DTO

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

archipy.models.dtos.fastapi_exception_response_dto.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",
                            },
                        },
                    },
                },
            },
        }

archipy.models.dtos.fastapi_exception_response_dto.FastAPIErrorResponseDTO.status_code instance-attribute 

status_code = exception.http_status

archipy.models.dtos.fastapi_exception_response_dto.FastAPIErrorResponseDTO.model instance-attribute 

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",
                    },
                },
            }
        }
    },
}

archipy.models.dtos.fastapi_exception_response_dto.ValidationErrorResponseDTO

Bases: FastAPIErrorResponseDTO

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",
                                    },
                                ],
                            },
                        },
                    },
                },
            },
        }

archipy.models.dtos.fastapi_exception_response_dto.ValidationErrorResponseDTO.status_code instance-attribute 

status_code = HTTPStatus.UNPROCESSABLE_ENTITY

archipy.models.dtos.fastapi_exception_response_dto.ValidationErrorResponseDTO.model instance-attribute 

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

RediSearch DTOs

DTOs for RediSearch index schemas, document upserts, queries, aggregations, and normalized search results.

archipy.models.dtos.redis.search.AggregationDTO

Bases: BaseDTO

RediSearch aggregation request parameters.

Source code in archipy/models/dtos/redis/search/aggregation_dto.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
class AggregationDTO(BaseDTO):
    """RediSearch aggregation request parameters."""

    query: str = "*"
    group_by: list[str] | None = None
    reduce_field: str | None = None
    reduce_function: str | None = None
    sort_by: str | None = None
    sort_direction: str = "ASC"
    limit: int | None = None

archipy.models.dtos.redis.search.AggregationDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.AggregationDTO.query class-attribute instance-attribute 

query: str = '*'

archipy.models.dtos.redis.search.AggregationDTO.group_by class-attribute instance-attribute 

group_by: list[str] | None = None

archipy.models.dtos.redis.search.AggregationDTO.reduce_field class-attribute instance-attribute 

reduce_field: str | None = None

archipy.models.dtos.redis.search.AggregationDTO.reduce_function class-attribute instance-attribute 

reduce_function: str | None = None

archipy.models.dtos.redis.search.AggregationDTO.sort_by class-attribute instance-attribute 

sort_by: str | None = None

archipy.models.dtos.redis.search.AggregationDTO.sort_direction class-attribute instance-attribute 

sort_direction: str = 'ASC'

archipy.models.dtos.redis.search.AggregationDTO.limit class-attribute instance-attribute 

limit: int | None = None

archipy.models.dtos.redis.search.HashDocumentUpsertDTO

Bases: BaseDTO

Payload for upserting a HASH-backed RediSearch document.

Source code in archipy/models/dtos/redis/search/document_dto.py 
 4
 5
 6
 7
 8
 9
10
class HashDocumentUpsertDTO(BaseDTO):
    """Payload for upserting a HASH-backed RediSearch document."""

    doc_id: str
    fields: dict[str, str | int | float]
    vector_field: str | None = None
    vector: list[float] | None = None

archipy.models.dtos.redis.search.HashDocumentUpsertDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.HashDocumentUpsertDTO.doc_id instance-attribute 

doc_id: str

archipy.models.dtos.redis.search.HashDocumentUpsertDTO.fields instance-attribute 

fields: dict[str, str | int | float]

archipy.models.dtos.redis.search.HashDocumentUpsertDTO.vector_field class-attribute instance-attribute 

vector_field: str | None = None

archipy.models.dtos.redis.search.HashDocumentUpsertDTO.vector class-attribute instance-attribute 

vector: list[float] | None = None

archipy.models.dtos.redis.search.JsonDocumentUpsertDTO

Bases: BaseDTO

Payload for upserting a JSON-backed RediSearch document.

Source code in archipy/models/dtos/redis/search/document_dto.py 
13
14
15
16
17
18
class JsonDocumentUpsertDTO(BaseDTO):
    """Payload for upserting a JSON-backed RediSearch document."""

    doc_id: str
    payload: dict[str, str | int | float | list[float]]
    json_path: str = "$"

archipy.models.dtos.redis.search.JsonDocumentUpsertDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.JsonDocumentUpsertDTO.doc_id instance-attribute 

doc_id: str

archipy.models.dtos.redis.search.JsonDocumentUpsertDTO.payload instance-attribute 

payload: dict[str, str | int | float | list[float]]

archipy.models.dtos.redis.search.JsonDocumentUpsertDTO.json_path class-attribute instance-attribute 

json_path: str = '$'

archipy.models.dtos.redis.search.IndexSchemaDTO

Bases: BaseDTO

Schema describing fields indexed by RediSearch.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
 97
 98
 99
100
101
class IndexSchemaDTO(BaseDTO):
    """Schema describing fields indexed by RediSearch."""

    fields: list[IndexFieldConfig]
    index_type: RedisIndexType = RedisIndexType.HASH

archipy.models.dtos.redis.search.IndexSchemaDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.IndexSchemaDTO.fields instance-attribute 

fields: list[IndexFieldConfig]

archipy.models.dtos.redis.search.IndexSchemaDTO.index_type class-attribute instance-attribute 

index_type: RedisIndexType = RedisIndexType.HASH

archipy.models.dtos.redis.search.NumericFieldConfig

Bases: BaseDTO

RediSearch numeric field definition.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
30
31
32
33
34
class NumericFieldConfig(BaseDTO):
    """RediSearch numeric field definition."""

    field_type: Literal["numeric"] = "numeric"
    name: str

archipy.models.dtos.redis.search.NumericFieldConfig.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.NumericFieldConfig.field_type class-attribute instance-attribute 

field_type: Literal['numeric'] = 'numeric'

archipy.models.dtos.redis.search.NumericFieldConfig.name instance-attribute 

name: str

archipy.models.dtos.redis.search.TagFieldConfig

Bases: BaseDTO

RediSearch tag field definition.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
22
23
24
25
26
27
class TagFieldConfig(BaseDTO):
    """RediSearch tag field definition."""

    field_type: Literal["tag"] = "tag"
    name: str
    separator: str | None = None

archipy.models.dtos.redis.search.TagFieldConfig.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.TagFieldConfig.field_type class-attribute instance-attribute 

field_type: Literal['tag'] = 'tag'

archipy.models.dtos.redis.search.TagFieldConfig.name instance-attribute 

name: str

archipy.models.dtos.redis.search.TagFieldConfig.separator class-attribute instance-attribute 

separator: str | None = None

archipy.models.dtos.redis.search.TextFieldConfig

Bases: BaseDTO

RediSearch text field definition.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
15
16
17
18
19
class TextFieldConfig(BaseDTO):
    """RediSearch text field definition."""

    field_type: Literal["text"] = "text"
    name: str

archipy.models.dtos.redis.search.TextFieldConfig.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.TextFieldConfig.field_type class-attribute instance-attribute 

field_type: Literal['text'] = 'text'

archipy.models.dtos.redis.search.TextFieldConfig.name instance-attribute 

name: str

archipy.models.dtos.redis.search.VectorFieldConfig

Bases: BaseDTO

RediSearch vector field definition.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
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
80
81
82
83
84
85
86
87
88
89
90
91
class VectorFieldConfig(BaseDTO):
    """RediSearch vector field definition."""

    field_type: Literal["vector"] = "vector"
    name: str
    dim: int
    distance_metric: VectorDistanceMetric = VectorDistanceMetric.COSINE
    algorithm: VectorAlgorithm = VectorAlgorithm.HNSW
    vector_type: VectorType = VectorType.FLOAT32
    m: int | None = None
    ef_construction: int | None = None
    ef_runtime: int | None = None
    epsilon: float | None = None
    compression: VectorCompression | None = None
    construction_window_size: int | None = None
    graph_max_degree: int | None = None
    search_window_size: int | None = None
    training_threshold: int | None = None
    reduce: int | None = None

    @model_validator(mode="after")
    def validate_algorithm_attributes(self) -> VectorFieldConfig:
        """Reject tuning attributes that do not apply to the selected algorithm."""
        hnsw_only = {
            "m": self.m,
            "ef_construction": self.ef_construction,
            "ef_runtime": self.ef_runtime,
        }
        svs_only = {
            "compression": self.compression,
            "construction_window_size": self.construction_window_size,
            "graph_max_degree": self.graph_max_degree,
            "search_window_size": self.search_window_size,
            "training_threshold": self.training_threshold,
            "reduce": self.reduce,
        }
        if self.algorithm == VectorAlgorithm.FLAT:
            invalid = [name for name, value in {**hnsw_only, **svs_only}.items() if value is not None]
            if invalid:
                msg = f"FLAT indexes do not support tuning attributes: {', '.join(invalid)}"
                raise ValueError(msg)
        if self.algorithm == VectorAlgorithm.HNSW:
            invalid = [name for name, value in svs_only.items() if value is not None]
            if invalid:
                msg = f"HNSW indexes do not support SVS-VAMANA attributes: {', '.join(invalid)}"
                raise ValueError(msg)
        if self.algorithm == VectorAlgorithm.SVS_VAMANA:
            invalid = [name for name, value in hnsw_only.items() if value is not None]
            if invalid:
                msg = f"SVS-VAMANA indexes do not support HNSW attributes: {', '.join(invalid)}"
                raise ValueError(msg)
            if self.vector_type not in {VectorType.FLOAT16, VectorType.FLOAT32}:
                msg = "SVS-VAMANA indexes support FLOAT16 and FLOAT32 vector types only"
                raise ValueError(msg)
        return self

archipy.models.dtos.redis.search.VectorFieldConfig.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.VectorFieldConfig.field_type class-attribute instance-attribute 

field_type: Literal['vector'] = 'vector'

archipy.models.dtos.redis.search.VectorFieldConfig.name instance-attribute 

name: str

archipy.models.dtos.redis.search.VectorFieldConfig.dim instance-attribute 

dim: int

archipy.models.dtos.redis.search.VectorFieldConfig.distance_metric class-attribute instance-attribute 

distance_metric: VectorDistanceMetric = (
    VectorDistanceMetric.COSINE
)

archipy.models.dtos.redis.search.VectorFieldConfig.algorithm class-attribute instance-attribute 

algorithm: VectorAlgorithm = VectorAlgorithm.HNSW

archipy.models.dtos.redis.search.VectorFieldConfig.vector_type class-attribute instance-attribute 

vector_type: VectorType = VectorType.FLOAT32

archipy.models.dtos.redis.search.VectorFieldConfig.m class-attribute instance-attribute 

m: int | None = None

archipy.models.dtos.redis.search.VectorFieldConfig.ef_construction class-attribute instance-attribute 

ef_construction: int | None = None

archipy.models.dtos.redis.search.VectorFieldConfig.ef_runtime class-attribute instance-attribute 

ef_runtime: int | None = None

archipy.models.dtos.redis.search.VectorFieldConfig.epsilon class-attribute instance-attribute 

epsilon: float | None = None

archipy.models.dtos.redis.search.VectorFieldConfig.compression class-attribute instance-attribute 

compression: VectorCompression | None = None

archipy.models.dtos.redis.search.VectorFieldConfig.construction_window_size class-attribute instance-attribute 

construction_window_size: int | None = None

archipy.models.dtos.redis.search.VectorFieldConfig.graph_max_degree class-attribute instance-attribute 

graph_max_degree: int | None = None

archipy.models.dtos.redis.search.VectorFieldConfig.search_window_size class-attribute instance-attribute 

search_window_size: int | None = None

archipy.models.dtos.redis.search.VectorFieldConfig.training_threshold class-attribute instance-attribute 

training_threshold: int | None = None

archipy.models.dtos.redis.search.VectorFieldConfig.reduce class-attribute instance-attribute 

reduce: int | None = None

archipy.models.dtos.redis.search.VectorFieldConfig.validate_algorithm_attributes 

validate_algorithm_attributes() -> VectorFieldConfig

Reject tuning attributes that do not apply to the selected algorithm.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
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
82
83
84
85
86
87
88
89
90
91
@model_validator(mode="after")
def validate_algorithm_attributes(self) -> VectorFieldConfig:
    """Reject tuning attributes that do not apply to the selected algorithm."""
    hnsw_only = {
        "m": self.m,
        "ef_construction": self.ef_construction,
        "ef_runtime": self.ef_runtime,
    }
    svs_only = {
        "compression": self.compression,
        "construction_window_size": self.construction_window_size,
        "graph_max_degree": self.graph_max_degree,
        "search_window_size": self.search_window_size,
        "training_threshold": self.training_threshold,
        "reduce": self.reduce,
    }
    if self.algorithm == VectorAlgorithm.FLAT:
        invalid = [name for name, value in {**hnsw_only, **svs_only}.items() if value is not None]
        if invalid:
            msg = f"FLAT indexes do not support tuning attributes: {', '.join(invalid)}"
            raise ValueError(msg)
    if self.algorithm == VectorAlgorithm.HNSW:
        invalid = [name for name, value in svs_only.items() if value is not None]
        if invalid:
            msg = f"HNSW indexes do not support SVS-VAMANA attributes: {', '.join(invalid)}"
            raise ValueError(msg)
    if self.algorithm == VectorAlgorithm.SVS_VAMANA:
        invalid = [name for name, value in hnsw_only.items() if value is not None]
        if invalid:
            msg = f"SVS-VAMANA indexes do not support HNSW attributes: {', '.join(invalid)}"
            raise ValueError(msg)
        if self.vector_type not in {VectorType.FLOAT16, VectorType.FLOAT32}:
            msg = "SVS-VAMANA indexes support FLOAT16 and FLOAT32 vector types only"
            raise ValueError(msg)
    return self

archipy.models.dtos.redis.search.KnnQueryDTO

Bases: BaseDTO

Vector KNN search parameters.

Source code in archipy/models/dtos/redis/search/search_query_dto.py 
18
19
20
21
22
23
24
25
26
27
class KnnQueryDTO(BaseDTO):
    """Vector KNN search parameters."""

    vector: list[float]
    vector_field: str = "embedding"
    k: int = 10
    filter_expr: str | None = None
    return_fields: list[str] | None = None
    score_field: str = "score"
    runtime: VectorQueryRuntimeDTO | None = None

archipy.models.dtos.redis.search.KnnQueryDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.KnnQueryDTO.vector instance-attribute 

vector: list[float]

archipy.models.dtos.redis.search.KnnQueryDTO.vector_field class-attribute instance-attribute 

vector_field: str = 'embedding'

archipy.models.dtos.redis.search.KnnQueryDTO.k class-attribute instance-attribute 

k: int = 10

archipy.models.dtos.redis.search.KnnQueryDTO.filter_expr class-attribute instance-attribute 

filter_expr: str | None = None

archipy.models.dtos.redis.search.KnnQueryDTO.return_fields class-attribute instance-attribute 

return_fields: list[str] | None = None

archipy.models.dtos.redis.search.KnnQueryDTO.score_field class-attribute instance-attribute 

score_field: str = 'score'

archipy.models.dtos.redis.search.KnnQueryDTO.runtime class-attribute instance-attribute 

runtime: VectorQueryRuntimeDTO | None = None

archipy.models.dtos.redis.search.SearchQueryDTO

Bases: BaseDTO

Full-text, vector KNN, vector range, or hybrid RediSearch query.

Source code in archipy/models/dtos/redis/search/search_query_dto.py 
 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
 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
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
class SearchQueryDTO(BaseDTO):
    """Full-text, vector KNN, vector range, or hybrid RediSearch query."""

    query: str = "*"
    return_fields: list[str] | None = None
    offset: int = 0
    limit: int = 10
    text_scorer: str | None = None
    knn: KnnQueryDTO | None = None
    range: RangeQueryDTO | None = None

    @property
    def is_hybrid(self) -> bool:
        """Return whether the query combines full-text and vector KNN."""
        return self.knn is not None and self.query != "*"

    @property
    def is_range(self) -> bool:
        """Return whether the query is a vector range search."""
        return self.range is not None

    @classmethod
    def from_knn(
        cls,
        vector: list[float],
        *,
        k: int = 10,
        vector_field: str = "embedding",
        filter_expr: str | None = None,
        return_fields: list[str] | None = None,
        runtime: VectorQueryRuntimeDTO | None = None,
    ) -> SearchQueryDTO:
        """Build a KNN-only search query.

        Args:
            vector: Query embedding vector.
            k: Number of nearest neighbors to return.
            vector_field: Indexed vector field name.
            filter_expr: Optional RediSearch filter expression.
            return_fields: Optional document fields to return.
            runtime: Optional vector query runtime parameters.

        Returns:
            SearchQueryDTO configured for vector KNN search.
        """
        return cls(
            knn=KnnQueryDTO(
                vector=vector,
                k=k,
                vector_field=vector_field,
                filter_expr=filter_expr,
                return_fields=return_fields,
                runtime=runtime,
            ),
        )

    @classmethod
    def from_range(
        cls,
        vector: list[float],
        *,
        radius: float,
        vector_field: str = "embedding",
        filter_expr: str | None = None,
        return_fields: list[str] | None = None,
        score_field: str | None = None,
        limit: int = 10,
        runtime: VectorQueryRuntimeDTO | None = None,
    ) -> SearchQueryDTO:
        """Build a vector range search query.

        Args:
            vector: Query embedding vector.
            radius: Maximum semantic distance from the query vector.
            vector_field: Indexed vector field name.
            filter_expr: Optional RediSearch filter expression combined with range.
            return_fields: Optional document fields to return.
            score_field: Optional distance field name in the response.
            limit: Maximum number of documents to return.
            runtime: Optional vector query runtime parameters.

        Returns:
            SearchQueryDTO configured for vector range search.
        """
        return cls(
            limit=limit,
            range=RangeQueryDTO(
                vector=vector,
                radius=radius,
                vector_field=vector_field,
                filter_expr=filter_expr,
                return_fields=return_fields,
                score_field=score_field,
                runtime=runtime,
            ),
        )

    @classmethod
    def from_hybrid(
        cls,
        text_query: str,
        vector: list[float],
        *,
        k: int = 10,
        vector_field: str = "embedding",
        filter_expr: str | None = None,
        text_scorer: str | None = None,
        return_fields: list[str] | None = None,
        runtime: VectorQueryRuntimeDTO | None = None,
    ) -> SearchQueryDTO:
        """Build a hybrid full-text and vector search query.

        Args:
            text_query: Full-text query string.
            vector: Query embedding vector.
            k: Number of nearest neighbors to return.
            vector_field: Indexed vector field name.
            filter_expr: Optional RediSearch filter expression.
            text_scorer: Optional full-text scorer name.
            return_fields: Optional document fields to return.
            runtime: Optional vector query runtime parameters.

        Returns:
            SearchQueryDTO configured for hybrid search.
        """
        return cls(
            query=text_query,
            text_scorer=text_scorer,
            knn=KnnQueryDTO(
                vector=vector,
                k=k,
                vector_field=vector_field,
                filter_expr=filter_expr,
                return_fields=return_fields,
                runtime=runtime,
            ),
        )

archipy.models.dtos.redis.search.SearchQueryDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.SearchQueryDTO.query class-attribute instance-attribute 

query: str = '*'

archipy.models.dtos.redis.search.SearchQueryDTO.return_fields class-attribute instance-attribute 

return_fields: list[str] | None = None

archipy.models.dtos.redis.search.SearchQueryDTO.offset class-attribute instance-attribute 

offset: int = 0

archipy.models.dtos.redis.search.SearchQueryDTO.limit class-attribute instance-attribute 

limit: int = 10

archipy.models.dtos.redis.search.SearchQueryDTO.text_scorer class-attribute instance-attribute 

text_scorer: str | None = None

archipy.models.dtos.redis.search.SearchQueryDTO.knn class-attribute instance-attribute 

knn: KnnQueryDTO | None = None

archipy.models.dtos.redis.search.SearchQueryDTO.range class-attribute instance-attribute 

range: RangeQueryDTO | None = None

archipy.models.dtos.redis.search.SearchQueryDTO.is_hybrid property 

is_hybrid: bool

Return whether the query combines full-text and vector KNN.

archipy.models.dtos.redis.search.SearchQueryDTO.is_range property 

is_range: bool

Return whether the query is a vector range search.

archipy.models.dtos.redis.search.SearchQueryDTO.from_knn classmethod 

from_knn(
    vector: list[float],
    *,
    k: int = 10,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    return_fields: list[str] | None = None,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO

Build a KNN-only search query.

Parameters:

Name Type Description Default
vector list[float]

Query embedding vector.

 required
k int

Number of nearest neighbors to return.

 10
vector_field str

Indexed vector field name.

 'embedding'
filter_expr str | None

Optional RediSearch filter expression.

 None
return_fields list[str] | None

Optional document fields to return.

 None
runtime VectorQueryRuntimeDTO | None

Optional vector query runtime parameters.

 None

Returns:

Type Description
SearchQueryDTO

SearchQueryDTO configured for vector KNN search.
Source code in archipy/models/dtos/redis/search/search_query_dto.py 
63
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
@classmethod
def from_knn(
    cls,
    vector: list[float],
    *,
    k: int = 10,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    return_fields: list[str] | None = None,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO:
    """Build a KNN-only search query.

    Args:
        vector: Query embedding vector.
        k: Number of nearest neighbors to return.
        vector_field: Indexed vector field name.
        filter_expr: Optional RediSearch filter expression.
        return_fields: Optional document fields to return.
        runtime: Optional vector query runtime parameters.

    Returns:
        SearchQueryDTO configured for vector KNN search.
    """
    return cls(
        knn=KnnQueryDTO(
            vector=vector,
            k=k,
            vector_field=vector_field,
            filter_expr=filter_expr,
            return_fields=return_fields,
            runtime=runtime,
        ),
    )

archipy.models.dtos.redis.search.SearchQueryDTO.from_range classmethod 

from_range(
    vector: list[float],
    *,
    radius: float,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    return_fields: list[str] | None = None,
    score_field: str | None = None,
    limit: int = 10,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO

Build a vector range search query.

Parameters:

Name Type Description Default
vector list[float]

Query embedding vector.

 required
radius float

Maximum semantic distance from the query vector.

 required
vector_field str

Indexed vector field name.

 'embedding'
filter_expr str | None

Optional RediSearch filter expression combined with range.

 None
return_fields list[str] | None

Optional document fields to return.

 None
score_field str | None

Optional distance field name in the response.

 None
limit int

Maximum number of documents to return.

 10
runtime VectorQueryRuntimeDTO | None

Optional vector query runtime parameters.

 None

Returns:

Type Description
SearchQueryDTO

SearchQueryDTO configured for vector range search.
Source code in archipy/models/dtos/redis/search/search_query_dto.py 
 98
 99
100
101
102
103
104
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
@classmethod
def from_range(
    cls,
    vector: list[float],
    *,
    radius: float,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    return_fields: list[str] | None = None,
    score_field: str | None = None,
    limit: int = 10,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO:
    """Build a vector range search query.

    Args:
        vector: Query embedding vector.
        radius: Maximum semantic distance from the query vector.
        vector_field: Indexed vector field name.
        filter_expr: Optional RediSearch filter expression combined with range.
        return_fields: Optional document fields to return.
        score_field: Optional distance field name in the response.
        limit: Maximum number of documents to return.
        runtime: Optional vector query runtime parameters.

    Returns:
        SearchQueryDTO configured for vector range search.
    """
    return cls(
        limit=limit,
        range=RangeQueryDTO(
            vector=vector,
            radius=radius,
            vector_field=vector_field,
            filter_expr=filter_expr,
            return_fields=return_fields,
            score_field=score_field,
            runtime=runtime,
        ),
    )

archipy.models.dtos.redis.search.SearchQueryDTO.from_hybrid classmethod 

from_hybrid(
    text_query: str,
    vector: list[float],
    *,
    k: int = 10,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    text_scorer: str | None = None,
    return_fields: list[str] | None = None,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO

Build a hybrid full-text and vector search query.

Parameters:

Name Type Description Default
text_query str

Full-text query string.

 required
vector list[float]

Query embedding vector.

 required
k int

Number of nearest neighbors to return.

 10
vector_field str

Indexed vector field name.

 'embedding'
filter_expr str | None

Optional RediSearch filter expression.

 None
text_scorer str | None

Optional full-text scorer name.

 None
return_fields list[str] | None

Optional document fields to return.

 None
runtime VectorQueryRuntimeDTO | None

Optional vector query runtime parameters.

 None

Returns:

Type Description
SearchQueryDTO

SearchQueryDTO configured for hybrid search.
Source code in archipy/models/dtos/redis/search/search_query_dto.py 
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
@classmethod
def from_hybrid(
    cls,
    text_query: str,
    vector: list[float],
    *,
    k: int = 10,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    text_scorer: str | None = None,
    return_fields: list[str] | None = None,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO:
    """Build a hybrid full-text and vector search query.

    Args:
        text_query: Full-text query string.
        vector: Query embedding vector.
        k: Number of nearest neighbors to return.
        vector_field: Indexed vector field name.
        filter_expr: Optional RediSearch filter expression.
        text_scorer: Optional full-text scorer name.
        return_fields: Optional document fields to return.
        runtime: Optional vector query runtime parameters.

    Returns:
        SearchQueryDTO configured for hybrid search.
    """
    return cls(
        query=text_query,
        text_scorer=text_scorer,
        knn=KnnQueryDTO(
            vector=vector,
            k=k,
            vector_field=vector_field,
            filter_expr=filter_expr,
            return_fields=return_fields,
            runtime=runtime,
        ),
    )

archipy.models.dtos.redis.search.SearchHitDTO

Bases: BaseDTO

Single RediSearch document hit.

Source code in archipy/models/dtos/redis/search/search_result_dto.py 
 6
 7
 8
 9
10
11
class SearchHitDTO(BaseDTO):
    """Single RediSearch document hit."""

    doc_id: str
    score: float | None = None
    fields: dict[str, str | int | float | list[float] | bytes] = Field(default_factory=dict)

archipy.models.dtos.redis.search.SearchHitDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.SearchHitDTO.doc_id instance-attribute 

doc_id: str

archipy.models.dtos.redis.search.SearchHitDTO.score class-attribute instance-attribute 

score: float | None = None

archipy.models.dtos.redis.search.SearchHitDTO.fields class-attribute instance-attribute 

fields: dict[
    str, str | int | float | list[float] | bytes
] = Field(default_factory=dict)

archipy.models.dtos.redis.search.SearchResultDTO

Bases: BaseDTO

Normalized RediSearch query result.

Source code in archipy/models/dtos/redis/search/search_result_dto.py 
14
15
16
17
18
19
20
class SearchResultDTO(BaseDTO):
    """Normalized RediSearch query result."""

    total: int
    hits: list[SearchHitDTO]
    duration_ms: float | None = None
    warnings: list[str] = Field(default_factory=list)

archipy.models.dtos.redis.search.SearchResultDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.SearchResultDTO.total instance-attribute 

total: int

archipy.models.dtos.redis.search.SearchResultDTO.hits instance-attribute 

hits: list[SearchHitDTO]

archipy.models.dtos.redis.search.SearchResultDTO.duration_ms class-attribute instance-attribute 

duration_ms: float | None = None

archipy.models.dtos.redis.search.SearchResultDTO.warnings class-attribute instance-attribute 

warnings: list[str] = Field(default_factory=list)

archipy.models.dtos.redis.search.aggregation_dto

archipy.models.dtos.redis.search.aggregation_dto.AggregationDTO

Bases: BaseDTO

RediSearch aggregation request parameters.

Source code in archipy/models/dtos/redis/search/aggregation_dto.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
class AggregationDTO(BaseDTO):
    """RediSearch aggregation request parameters."""

    query: str = "*"
    group_by: list[str] | None = None
    reduce_field: str | None = None
    reduce_function: str | None = None
    sort_by: str | None = None
    sort_direction: str = "ASC"
    limit: int | None = None

archipy.models.dtos.redis.search.aggregation_dto.AggregationDTO.query class-attribute instance-attribute 

query: str = '*'

archipy.models.dtos.redis.search.aggregation_dto.AggregationDTO.group_by class-attribute instance-attribute 

group_by: list[str] | None = None

archipy.models.dtos.redis.search.aggregation_dto.AggregationDTO.reduce_field class-attribute instance-attribute 

reduce_field: str | None = None

archipy.models.dtos.redis.search.aggregation_dto.AggregationDTO.reduce_function class-attribute instance-attribute 

reduce_function: str | None = None

archipy.models.dtos.redis.search.aggregation_dto.AggregationDTO.sort_by class-attribute instance-attribute 

sort_by: str | None = None

archipy.models.dtos.redis.search.aggregation_dto.AggregationDTO.sort_direction class-attribute instance-attribute 

sort_direction: str = 'ASC'

archipy.models.dtos.redis.search.aggregation_dto.AggregationDTO.limit class-attribute instance-attribute 

limit: int | None = None

archipy.models.dtos.redis.search.aggregation_dto.AggregationDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.document_dto

archipy.models.dtos.redis.search.document_dto.HashDocumentUpsertDTO

Bases: BaseDTO

Payload for upserting a HASH-backed RediSearch document.

Source code in archipy/models/dtos/redis/search/document_dto.py 
 4
 5
 6
 7
 8
 9
10
class HashDocumentUpsertDTO(BaseDTO):
    """Payload for upserting a HASH-backed RediSearch document."""

    doc_id: str
    fields: dict[str, str | int | float]
    vector_field: str | None = None
    vector: list[float] | None = None

archipy.models.dtos.redis.search.document_dto.HashDocumentUpsertDTO.doc_id instance-attribute 

doc_id: str

archipy.models.dtos.redis.search.document_dto.HashDocumentUpsertDTO.fields instance-attribute 

fields: dict[str, str | int | float]

archipy.models.dtos.redis.search.document_dto.HashDocumentUpsertDTO.vector_field class-attribute instance-attribute 

vector_field: str | None = None

archipy.models.dtos.redis.search.document_dto.HashDocumentUpsertDTO.vector class-attribute instance-attribute 

vector: list[float] | None = None

archipy.models.dtos.redis.search.document_dto.HashDocumentUpsertDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.document_dto.JsonDocumentUpsertDTO

Bases: BaseDTO

Payload for upserting a JSON-backed RediSearch document.

Source code in archipy/models/dtos/redis/search/document_dto.py 
13
14
15
16
17
18
class JsonDocumentUpsertDTO(BaseDTO):
    """Payload for upserting a JSON-backed RediSearch document."""

    doc_id: str
    payload: dict[str, str | int | float | list[float]]
    json_path: str = "$"

archipy.models.dtos.redis.search.document_dto.JsonDocumentUpsertDTO.doc_id instance-attribute 

doc_id: str

archipy.models.dtos.redis.search.document_dto.JsonDocumentUpsertDTO.payload instance-attribute 

payload: dict[str, str | int | float | list[float]]

archipy.models.dtos.redis.search.document_dto.JsonDocumentUpsertDTO.json_path class-attribute instance-attribute 

json_path: str = '$'

archipy.models.dtos.redis.search.document_dto.JsonDocumentUpsertDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.index_schema_dto

archipy.models.dtos.redis.search.index_schema_dto.IndexFieldConfig module-attribute 

IndexFieldConfig = (
    TextFieldConfig
    | TagFieldConfig
    | NumericFieldConfig
    | VectorFieldConfig
)

archipy.models.dtos.redis.search.index_schema_dto.TextFieldConfig

Bases: BaseDTO

RediSearch text field definition.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
15
16
17
18
19
class TextFieldConfig(BaseDTO):
    """RediSearch text field definition."""

    field_type: Literal["text"] = "text"
    name: str

archipy.models.dtos.redis.search.index_schema_dto.TextFieldConfig.field_type class-attribute instance-attribute 

field_type: Literal['text'] = 'text'

archipy.models.dtos.redis.search.index_schema_dto.TextFieldConfig.name instance-attribute 

name: str

archipy.models.dtos.redis.search.index_schema_dto.TextFieldConfig.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.index_schema_dto.TagFieldConfig

Bases: BaseDTO

RediSearch tag field definition.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
22
23
24
25
26
27
class TagFieldConfig(BaseDTO):
    """RediSearch tag field definition."""

    field_type: Literal["tag"] = "tag"
    name: str
    separator: str | None = None

archipy.models.dtos.redis.search.index_schema_dto.TagFieldConfig.field_type class-attribute instance-attribute 

field_type: Literal['tag'] = 'tag'

archipy.models.dtos.redis.search.index_schema_dto.TagFieldConfig.name instance-attribute 

name: str

archipy.models.dtos.redis.search.index_schema_dto.TagFieldConfig.separator class-attribute instance-attribute 

separator: str | None = None

archipy.models.dtos.redis.search.index_schema_dto.TagFieldConfig.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.index_schema_dto.NumericFieldConfig

Bases: BaseDTO

RediSearch numeric field definition.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
30
31
32
33
34
class NumericFieldConfig(BaseDTO):
    """RediSearch numeric field definition."""

    field_type: Literal["numeric"] = "numeric"
    name: str

archipy.models.dtos.redis.search.index_schema_dto.NumericFieldConfig.field_type class-attribute instance-attribute 

field_type: Literal['numeric'] = 'numeric'

archipy.models.dtos.redis.search.index_schema_dto.NumericFieldConfig.name instance-attribute 

name: str

archipy.models.dtos.redis.search.index_schema_dto.NumericFieldConfig.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig

Bases: BaseDTO

RediSearch vector field definition.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
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
80
81
82
83
84
85
86
87
88
89
90
91
class VectorFieldConfig(BaseDTO):
    """RediSearch vector field definition."""

    field_type: Literal["vector"] = "vector"
    name: str
    dim: int
    distance_metric: VectorDistanceMetric = VectorDistanceMetric.COSINE
    algorithm: VectorAlgorithm = VectorAlgorithm.HNSW
    vector_type: VectorType = VectorType.FLOAT32
    m: int | None = None
    ef_construction: int | None = None
    ef_runtime: int | None = None
    epsilon: float | None = None
    compression: VectorCompression | None = None
    construction_window_size: int | None = None
    graph_max_degree: int | None = None
    search_window_size: int | None = None
    training_threshold: int | None = None
    reduce: int | None = None

    @model_validator(mode="after")
    def validate_algorithm_attributes(self) -> VectorFieldConfig:
        """Reject tuning attributes that do not apply to the selected algorithm."""
        hnsw_only = {
            "m": self.m,
            "ef_construction": self.ef_construction,
            "ef_runtime": self.ef_runtime,
        }
        svs_only = {
            "compression": self.compression,
            "construction_window_size": self.construction_window_size,
            "graph_max_degree": self.graph_max_degree,
            "search_window_size": self.search_window_size,
            "training_threshold": self.training_threshold,
            "reduce": self.reduce,
        }
        if self.algorithm == VectorAlgorithm.FLAT:
            invalid = [name for name, value in {**hnsw_only, **svs_only}.items() if value is not None]
            if invalid:
                msg = f"FLAT indexes do not support tuning attributes: {', '.join(invalid)}"
                raise ValueError(msg)
        if self.algorithm == VectorAlgorithm.HNSW:
            invalid = [name for name, value in svs_only.items() if value is not None]
            if invalid:
                msg = f"HNSW indexes do not support SVS-VAMANA attributes: {', '.join(invalid)}"
                raise ValueError(msg)
        if self.algorithm == VectorAlgorithm.SVS_VAMANA:
            invalid = [name for name, value in hnsw_only.items() if value is not None]
            if invalid:
                msg = f"SVS-VAMANA indexes do not support HNSW attributes: {', '.join(invalid)}"
                raise ValueError(msg)
            if self.vector_type not in {VectorType.FLOAT16, VectorType.FLOAT32}:
                msg = "SVS-VAMANA indexes support FLOAT16 and FLOAT32 vector types only"
                raise ValueError(msg)
        return self

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.field_type class-attribute instance-attribute 

field_type: Literal['vector'] = 'vector'

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.name instance-attribute 

name: str

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.dim instance-attribute 

dim: int

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.distance_metric class-attribute instance-attribute 

distance_metric: VectorDistanceMetric = (
    VectorDistanceMetric.COSINE
)

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.algorithm class-attribute instance-attribute 

algorithm: VectorAlgorithm = VectorAlgorithm.HNSW

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.vector_type class-attribute instance-attribute 

vector_type: VectorType = VectorType.FLOAT32

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.m class-attribute instance-attribute 

m: int | None = None

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.ef_construction class-attribute instance-attribute 

ef_construction: int | None = None

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.ef_runtime class-attribute instance-attribute 

ef_runtime: int | None = None

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.epsilon class-attribute instance-attribute 

epsilon: float | None = None

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.compression class-attribute instance-attribute 

compression: VectorCompression | None = None

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.construction_window_size class-attribute instance-attribute 

construction_window_size: int | None = None

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.graph_max_degree class-attribute instance-attribute 

graph_max_degree: int | None = None

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.search_window_size class-attribute instance-attribute 

search_window_size: int | None = None

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.training_threshold class-attribute instance-attribute 

training_threshold: int | None = None

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.reduce class-attribute instance-attribute 

reduce: int | None = None

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.index_schema_dto.VectorFieldConfig.validate_algorithm_attributes 

validate_algorithm_attributes() -> VectorFieldConfig

Reject tuning attributes that do not apply to the selected algorithm.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
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
82
83
84
85
86
87
88
89
90
91
@model_validator(mode="after")
def validate_algorithm_attributes(self) -> VectorFieldConfig:
    """Reject tuning attributes that do not apply to the selected algorithm."""
    hnsw_only = {
        "m": self.m,
        "ef_construction": self.ef_construction,
        "ef_runtime": self.ef_runtime,
    }
    svs_only = {
        "compression": self.compression,
        "construction_window_size": self.construction_window_size,
        "graph_max_degree": self.graph_max_degree,
        "search_window_size": self.search_window_size,
        "training_threshold": self.training_threshold,
        "reduce": self.reduce,
    }
    if self.algorithm == VectorAlgorithm.FLAT:
        invalid = [name for name, value in {**hnsw_only, **svs_only}.items() if value is not None]
        if invalid:
            msg = f"FLAT indexes do not support tuning attributes: {', '.join(invalid)}"
            raise ValueError(msg)
    if self.algorithm == VectorAlgorithm.HNSW:
        invalid = [name for name, value in svs_only.items() if value is not None]
        if invalid:
            msg = f"HNSW indexes do not support SVS-VAMANA attributes: {', '.join(invalid)}"
            raise ValueError(msg)
    if self.algorithm == VectorAlgorithm.SVS_VAMANA:
        invalid = [name for name, value in hnsw_only.items() if value is not None]
        if invalid:
            msg = f"SVS-VAMANA indexes do not support HNSW attributes: {', '.join(invalid)}"
            raise ValueError(msg)
        if self.vector_type not in {VectorType.FLOAT16, VectorType.FLOAT32}:
            msg = "SVS-VAMANA indexes support FLOAT16 and FLOAT32 vector types only"
            raise ValueError(msg)
    return self

archipy.models.dtos.redis.search.index_schema_dto.IndexSchemaDTO

Bases: BaseDTO

Schema describing fields indexed by RediSearch.

Source code in archipy/models/dtos/redis/search/index_schema_dto.py 
 97
 98
 99
100
101
class IndexSchemaDTO(BaseDTO):
    """Schema describing fields indexed by RediSearch."""

    fields: list[IndexFieldConfig]
    index_type: RedisIndexType = RedisIndexType.HASH

archipy.models.dtos.redis.search.index_schema_dto.IndexSchemaDTO.fields instance-attribute 

fields: list[IndexFieldConfig]

archipy.models.dtos.redis.search.index_schema_dto.IndexSchemaDTO.index_type class-attribute instance-attribute 

index_type: RedisIndexType = RedisIndexType.HASH

archipy.models.dtos.redis.search.index_schema_dto.IndexSchemaDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.search_query_dto

archipy.models.dtos.redis.search.search_query_dto.VectorQueryRuntimeDTO

Bases: BaseDTO

Runtime tuning parameters for vector KNN and range queries.

Source code in archipy/models/dtos/redis/search/search_query_dto.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
class VectorQueryRuntimeDTO(BaseDTO):
    """Runtime tuning parameters for vector KNN and range queries."""

    ef_runtime: int | None = None
    epsilon: float | None = None
    hybrid_policy: VectorHybridPolicy | None = None
    batch_size: int | None = None
    shard_k_ratio: float | None = None
    search_window_size: int | None = None
    use_search_history: UseSearchHistory | None = None
    search_buffer_capacity: int | None = None

archipy.models.dtos.redis.search.search_query_dto.VectorQueryRuntimeDTO.ef_runtime class-attribute instance-attribute 

ef_runtime: int | None = None

archipy.models.dtos.redis.search.search_query_dto.VectorQueryRuntimeDTO.epsilon class-attribute instance-attribute 

epsilon: float | None = None

archipy.models.dtos.redis.search.search_query_dto.VectorQueryRuntimeDTO.hybrid_policy class-attribute instance-attribute 

hybrid_policy: VectorHybridPolicy | None = None

archipy.models.dtos.redis.search.search_query_dto.VectorQueryRuntimeDTO.batch_size class-attribute instance-attribute 

batch_size: int | None = None

archipy.models.dtos.redis.search.search_query_dto.VectorQueryRuntimeDTO.shard_k_ratio class-attribute instance-attribute 

shard_k_ratio: float | None = None

archipy.models.dtos.redis.search.search_query_dto.VectorQueryRuntimeDTO.search_window_size class-attribute instance-attribute 

search_window_size: int | None = None

archipy.models.dtos.redis.search.search_query_dto.VectorQueryRuntimeDTO.use_search_history class-attribute instance-attribute 

use_search_history: UseSearchHistory | None = None

archipy.models.dtos.redis.search.search_query_dto.VectorQueryRuntimeDTO.search_buffer_capacity class-attribute instance-attribute 

search_buffer_capacity: int | None = None

archipy.models.dtos.redis.search.search_query_dto.VectorQueryRuntimeDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.search_query_dto.KnnQueryDTO

Bases: BaseDTO

Vector KNN search parameters.

Source code in archipy/models/dtos/redis/search/search_query_dto.py 
18
19
20
21
22
23
24
25
26
27
class KnnQueryDTO(BaseDTO):
    """Vector KNN search parameters."""

    vector: list[float]
    vector_field: str = "embedding"
    k: int = 10
    filter_expr: str | None = None
    return_fields: list[str] | None = None
    score_field: str = "score"
    runtime: VectorQueryRuntimeDTO | None = None

archipy.models.dtos.redis.search.search_query_dto.KnnQueryDTO.vector instance-attribute 

vector: list[float]

archipy.models.dtos.redis.search.search_query_dto.KnnQueryDTO.vector_field class-attribute instance-attribute 

vector_field: str = 'embedding'

archipy.models.dtos.redis.search.search_query_dto.KnnQueryDTO.k class-attribute instance-attribute 

k: int = 10

archipy.models.dtos.redis.search.search_query_dto.KnnQueryDTO.filter_expr class-attribute instance-attribute 

filter_expr: str | None = None

archipy.models.dtos.redis.search.search_query_dto.KnnQueryDTO.return_fields class-attribute instance-attribute 

return_fields: list[str] | None = None

archipy.models.dtos.redis.search.search_query_dto.KnnQueryDTO.score_field class-attribute instance-attribute 

score_field: str = 'score'

archipy.models.dtos.redis.search.search_query_dto.KnnQueryDTO.runtime class-attribute instance-attribute 

runtime: VectorQueryRuntimeDTO | None = None

archipy.models.dtos.redis.search.search_query_dto.KnnQueryDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.search_query_dto.RangeQueryDTO

Bases: BaseDTO

Vector range search parameters.

Source code in archipy/models/dtos/redis/search/search_query_dto.py 
30
31
32
33
34
35
36
37
38
39
class RangeQueryDTO(BaseDTO):
    """Vector range search parameters."""

    vector: list[float]
    vector_field: str = "embedding"
    radius: float
    filter_expr: str | None = None
    return_fields: list[str] | None = None
    score_field: str | None = None
    runtime: VectorQueryRuntimeDTO | None = None

archipy.models.dtos.redis.search.search_query_dto.RangeQueryDTO.vector instance-attribute 

vector: list[float]

archipy.models.dtos.redis.search.search_query_dto.RangeQueryDTO.vector_field class-attribute instance-attribute 

vector_field: str = 'embedding'

archipy.models.dtos.redis.search.search_query_dto.RangeQueryDTO.radius instance-attribute 

radius: float

archipy.models.dtos.redis.search.search_query_dto.RangeQueryDTO.filter_expr class-attribute instance-attribute 

filter_expr: str | None = None

archipy.models.dtos.redis.search.search_query_dto.RangeQueryDTO.return_fields class-attribute instance-attribute 

return_fields: list[str] | None = None

archipy.models.dtos.redis.search.search_query_dto.RangeQueryDTO.score_field class-attribute instance-attribute 

score_field: str | None = None

archipy.models.dtos.redis.search.search_query_dto.RangeQueryDTO.runtime class-attribute instance-attribute 

runtime: VectorQueryRuntimeDTO | None = None

archipy.models.dtos.redis.search.search_query_dto.RangeQueryDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO

Bases: BaseDTO

Full-text, vector KNN, vector range, or hybrid RediSearch query.

Source code in archipy/models/dtos/redis/search/search_query_dto.py 
 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
 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
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
class SearchQueryDTO(BaseDTO):
    """Full-text, vector KNN, vector range, or hybrid RediSearch query."""

    query: str = "*"
    return_fields: list[str] | None = None
    offset: int = 0
    limit: int = 10
    text_scorer: str | None = None
    knn: KnnQueryDTO | None = None
    range: RangeQueryDTO | None = None

    @property
    def is_hybrid(self) -> bool:
        """Return whether the query combines full-text and vector KNN."""
        return self.knn is not None and self.query != "*"

    @property
    def is_range(self) -> bool:
        """Return whether the query is a vector range search."""
        return self.range is not None

    @classmethod
    def from_knn(
        cls,
        vector: list[float],
        *,
        k: int = 10,
        vector_field: str = "embedding",
        filter_expr: str | None = None,
        return_fields: list[str] | None = None,
        runtime: VectorQueryRuntimeDTO | None = None,
    ) -> SearchQueryDTO:
        """Build a KNN-only search query.

        Args:
            vector: Query embedding vector.
            k: Number of nearest neighbors to return.
            vector_field: Indexed vector field name.
            filter_expr: Optional RediSearch filter expression.
            return_fields: Optional document fields to return.
            runtime: Optional vector query runtime parameters.

        Returns:
            SearchQueryDTO configured for vector KNN search.
        """
        return cls(
            knn=KnnQueryDTO(
                vector=vector,
                k=k,
                vector_field=vector_field,
                filter_expr=filter_expr,
                return_fields=return_fields,
                runtime=runtime,
            ),
        )

    @classmethod
    def from_range(
        cls,
        vector: list[float],
        *,
        radius: float,
        vector_field: str = "embedding",
        filter_expr: str | None = None,
        return_fields: list[str] | None = None,
        score_field: str | None = None,
        limit: int = 10,
        runtime: VectorQueryRuntimeDTO | None = None,
    ) -> SearchQueryDTO:
        """Build a vector range search query.

        Args:
            vector: Query embedding vector.
            radius: Maximum semantic distance from the query vector.
            vector_field: Indexed vector field name.
            filter_expr: Optional RediSearch filter expression combined with range.
            return_fields: Optional document fields to return.
            score_field: Optional distance field name in the response.
            limit: Maximum number of documents to return.
            runtime: Optional vector query runtime parameters.

        Returns:
            SearchQueryDTO configured for vector range search.
        """
        return cls(
            limit=limit,
            range=RangeQueryDTO(
                vector=vector,
                radius=radius,
                vector_field=vector_field,
                filter_expr=filter_expr,
                return_fields=return_fields,
                score_field=score_field,
                runtime=runtime,
            ),
        )

    @classmethod
    def from_hybrid(
        cls,
        text_query: str,
        vector: list[float],
        *,
        k: int = 10,
        vector_field: str = "embedding",
        filter_expr: str | None = None,
        text_scorer: str | None = None,
        return_fields: list[str] | None = None,
        runtime: VectorQueryRuntimeDTO | None = None,
    ) -> SearchQueryDTO:
        """Build a hybrid full-text and vector search query.

        Args:
            text_query: Full-text query string.
            vector: Query embedding vector.
            k: Number of nearest neighbors to return.
            vector_field: Indexed vector field name.
            filter_expr: Optional RediSearch filter expression.
            text_scorer: Optional full-text scorer name.
            return_fields: Optional document fields to return.
            runtime: Optional vector query runtime parameters.

        Returns:
            SearchQueryDTO configured for hybrid search.
        """
        return cls(
            query=text_query,
            text_scorer=text_scorer,
            knn=KnnQueryDTO(
                vector=vector,
                k=k,
                vector_field=vector_field,
                filter_expr=filter_expr,
                return_fields=return_fields,
                runtime=runtime,
            ),
        )

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.query class-attribute instance-attribute 

query: str = '*'

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.return_fields class-attribute instance-attribute 

return_fields: list[str] | None = None

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.offset class-attribute instance-attribute 

offset: int = 0

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.limit class-attribute instance-attribute 

limit: int = 10

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.text_scorer class-attribute instance-attribute 

text_scorer: str | None = None

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.knn class-attribute instance-attribute 

knn: KnnQueryDTO | None = None

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.range class-attribute instance-attribute 

range: RangeQueryDTO | None = None

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.is_hybrid property 

is_hybrid: bool

Return whether the query combines full-text and vector KNN.

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.is_range property 

is_range: bool

Return whether the query is a vector range search.

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.from_knn classmethod 

from_knn(
    vector: list[float],
    *,
    k: int = 10,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    return_fields: list[str] | None = None,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO

Build a KNN-only search query.

Parameters:

Name Type Description Default
vector list[float]

Query embedding vector.

 required
k int

Number of nearest neighbors to return.

 10
vector_field str

Indexed vector field name.

 'embedding'
filter_expr str | None

Optional RediSearch filter expression.

 None
return_fields list[str] | None

Optional document fields to return.

 None
runtime VectorQueryRuntimeDTO | None

Optional vector query runtime parameters.

 None

Returns:

Type Description
SearchQueryDTO

SearchQueryDTO configured for vector KNN search.
Source code in archipy/models/dtos/redis/search/search_query_dto.py 
63
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
@classmethod
def from_knn(
    cls,
    vector: list[float],
    *,
    k: int = 10,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    return_fields: list[str] | None = None,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO:
    """Build a KNN-only search query.

    Args:
        vector: Query embedding vector.
        k: Number of nearest neighbors to return.
        vector_field: Indexed vector field name.
        filter_expr: Optional RediSearch filter expression.
        return_fields: Optional document fields to return.
        runtime: Optional vector query runtime parameters.

    Returns:
        SearchQueryDTO configured for vector KNN search.
    """
    return cls(
        knn=KnnQueryDTO(
            vector=vector,
            k=k,
            vector_field=vector_field,
            filter_expr=filter_expr,
            return_fields=return_fields,
            runtime=runtime,
        ),
    )

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.from_range classmethod 

from_range(
    vector: list[float],
    *,
    radius: float,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    return_fields: list[str] | None = None,
    score_field: str | None = None,
    limit: int = 10,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO

Build a vector range search query.

Parameters:

Name Type Description Default
vector list[float]

Query embedding vector.

 required
radius float

Maximum semantic distance from the query vector.

 required
vector_field str

Indexed vector field name.

 'embedding'
filter_expr str | None

Optional RediSearch filter expression combined with range.

 None
return_fields list[str] | None

Optional document fields to return.

 None
score_field str | None

Optional distance field name in the response.

 None
limit int

Maximum number of documents to return.

 10
runtime VectorQueryRuntimeDTO | None

Optional vector query runtime parameters.

 None

Returns:

Type Description
SearchQueryDTO

SearchQueryDTO configured for vector range search.
Source code in archipy/models/dtos/redis/search/search_query_dto.py 
 98
 99
100
101
102
103
104
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
@classmethod
def from_range(
    cls,
    vector: list[float],
    *,
    radius: float,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    return_fields: list[str] | None = None,
    score_field: str | None = None,
    limit: int = 10,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO:
    """Build a vector range search query.

    Args:
        vector: Query embedding vector.
        radius: Maximum semantic distance from the query vector.
        vector_field: Indexed vector field name.
        filter_expr: Optional RediSearch filter expression combined with range.
        return_fields: Optional document fields to return.
        score_field: Optional distance field name in the response.
        limit: Maximum number of documents to return.
        runtime: Optional vector query runtime parameters.

    Returns:
        SearchQueryDTO configured for vector range search.
    """
    return cls(
        limit=limit,
        range=RangeQueryDTO(
            vector=vector,
            radius=radius,
            vector_field=vector_field,
            filter_expr=filter_expr,
            return_fields=return_fields,
            score_field=score_field,
            runtime=runtime,
        ),
    )

archipy.models.dtos.redis.search.search_query_dto.SearchQueryDTO.from_hybrid classmethod 

from_hybrid(
    text_query: str,
    vector: list[float],
    *,
    k: int = 10,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    text_scorer: str | None = None,
    return_fields: list[str] | None = None,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO

Build a hybrid full-text and vector search query.

Parameters:

Name Type Description Default
text_query str

Full-text query string.

 required
vector list[float]

Query embedding vector.

 required
k int

Number of nearest neighbors to return.

 10
vector_field str

Indexed vector field name.

 'embedding'
filter_expr str | None

Optional RediSearch filter expression.

 None
text_scorer str | None

Optional full-text scorer name.

 None
return_fields list[str] | None

Optional document fields to return.

 None
runtime VectorQueryRuntimeDTO | None

Optional vector query runtime parameters.

 None

Returns:

Type Description
SearchQueryDTO

SearchQueryDTO configured for hybrid search.
Source code in archipy/models/dtos/redis/search/search_query_dto.py 
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
@classmethod
def from_hybrid(
    cls,
    text_query: str,
    vector: list[float],
    *,
    k: int = 10,
    vector_field: str = "embedding",
    filter_expr: str | None = None,
    text_scorer: str | None = None,
    return_fields: list[str] | None = None,
    runtime: VectorQueryRuntimeDTO | None = None,
) -> SearchQueryDTO:
    """Build a hybrid full-text and vector search query.

    Args:
        text_query: Full-text query string.
        vector: Query embedding vector.
        k: Number of nearest neighbors to return.
        vector_field: Indexed vector field name.
        filter_expr: Optional RediSearch filter expression.
        text_scorer: Optional full-text scorer name.
        return_fields: Optional document fields to return.
        runtime: Optional vector query runtime parameters.

    Returns:
        SearchQueryDTO configured for hybrid search.
    """
    return cls(
        query=text_query,
        text_scorer=text_scorer,
        knn=KnnQueryDTO(
            vector=vector,
            k=k,
            vector_field=vector_field,
            filter_expr=filter_expr,
            return_fields=return_fields,
            runtime=runtime,
        ),
    )

archipy.models.dtos.redis.search.search_result_dto

archipy.models.dtos.redis.search.search_result_dto.SearchHitDTO

Bases: BaseDTO

Single RediSearch document hit.

Source code in archipy/models/dtos/redis/search/search_result_dto.py 
 6
 7
 8
 9
10
11
class SearchHitDTO(BaseDTO):
    """Single RediSearch document hit."""

    doc_id: str
    score: float | None = None
    fields: dict[str, str | int | float | list[float] | bytes] = Field(default_factory=dict)

archipy.models.dtos.redis.search.search_result_dto.SearchHitDTO.doc_id instance-attribute 

doc_id: str

archipy.models.dtos.redis.search.search_result_dto.SearchHitDTO.score class-attribute instance-attribute 

score: float | None = None

archipy.models.dtos.redis.search.search_result_dto.SearchHitDTO.fields class-attribute instance-attribute 

fields: dict[
    str, str | int | float | list[float] | bytes
] = Field(default_factory=dict)

archipy.models.dtos.redis.search.search_result_dto.SearchHitDTO.model_config class-attribute instance-attribute 

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

archipy.models.dtos.redis.search.search_result_dto.SearchResultDTO

Bases: BaseDTO

Normalized RediSearch query result.

Source code in archipy/models/dtos/redis/search/search_result_dto.py 
14
15
16
17
18
19
20
class SearchResultDTO(BaseDTO):
    """Normalized RediSearch query result."""

    total: int
    hits: list[SearchHitDTO]
    duration_ms: float | None = None
    warnings: list[str] = Field(default_factory=list)

archipy.models.dtos.redis.search.search_result_dto.SearchResultDTO.total instance-attribute 

total: int

archipy.models.dtos.redis.search.search_result_dto.SearchResultDTO.hits instance-attribute 

hits: list[SearchHitDTO]

archipy.models.dtos.redis.search.search_result_dto.SearchResultDTO.duration_ms class-attribute instance-attribute 

duration_ms: float | None = None

archipy.models.dtos.redis.search.search_result_dto.SearchResultDTO.warnings class-attribute instance-attribute 

warnings: list[str] = Field(default_factory=list)

archipy.models.dtos.redis.search.search_result_dto.SearchResultDTO.model_config class-attribute instance-attribute 

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