Skip to content

Inner Types

kyotsu.config.databases._types

This module contains types that are used for kyotsu.config.databases mixins and models.

ATTRIBUTE DESCRIPTION
ConnType

TypeVar that used for generic DBConnection models.
CLASS DESCRIPTION
AmqpQueryParams

Represents the set of query params used for AMQP connection.
OpenSearchConnectionDict

Represents the dictionary of connection parameters needed for an OpenSearch connection.

ConnType = TypeVar('ConnType') module-attribute

Type for CONN_STR in classes inherited from generic DBConnection.

AmqpQueryParams

Bases: TypedDict

This TypedDict subclass represents the set of query parameters that can be used on an AMQP connection.

These parameters include detailed AMQP client configurations, such as backpressure detection, retry delay, socket timeout and many others.

This dict should be used as the QUERY_PARAMS attribute of an AmqpConnection instance.

See more details at: pika.connection.URLParameters

Source code in src/kyotsu/config/databases/_types.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class AmqpQueryParams(TypedDict, total=False):
    """
    This `TypedDict` subclass represents the set of query parameters that can be used on an AMQP connection.

    These parameters include detailed AMQP client configurations, such as backpressure detection, retry delay,
    socket timeout and many others.

    This dict should be used as the `QUERY_PARAMS` attribute of an `AmqpConnection` instance.

    See more details at: [pika.connection.URLParameters](https://pika.readthedocs.io/en/stable/modules/parameters.html#pika.connection.URLParameters)
    """

    backpressure_detection: Literal["t"]
    channel_max: int
    connection_attempts: int
    frame_max: int
    heartbeat: int
    locale: str
    ssl_options: dict[str, str]
    retry_delay: int
    socket_timeout: int

OpenSearchConnectionDict

Bases: TypedDict

This TypedDict subclass represents the dictionary of connection parameters needed for an OpenSearch connection.

It is used in the CONN_DICT attribute of an OpenSearchNoCertConnection instance after being processed in assemble_conn_dict method.

This dict establishes details about hosts, authentication, scheme, port, and SSL context, which all will be used to configure and establish the OpenSearch connection.

Source code in src/kyotsu/config/databases/_types.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class OpenSearchConnectionDict(TypedDict, total=False):
    """
    This `TypedDict` subclass represents the dictionary of connection parameters needed for an OpenSearch connection.

    It is used in the `CONN_DICT` attribute of an `OpenSearchNoCertConnection` instance after being processed in
    `assemble_conn_dict` method.

    This dict establishes details about hosts, authentication, scheme, port, and SSL context, which all will be used to
    configure and establish the OpenSearch connection.
    """

    hosts: Sequence[str]
    http_auth: tuple[str, str] | None
    scheme: Literal["http", "https"]
    port: str

    ssl_context: NotRequired[SSLContext]
    timeout: NotRequired[int]
    max_retries: NotRequired[int]
    retry_on_timeout: NotRequired[bool]