Skip to content

Configs

Overview

The configs module provides tools for standardised configuration management and injection, supporting consistent setup across services like databases, Redis, and email.

Quick Start

from archipy.configs.base_config import BaseConfig

class AppConfig(BaseConfig):
    APP_NAME: str = "MyService"
    DEBUG: bool = False
    DB_HOST: str = "localhost"
    DB_PORT: int = 5432

API Stability

Component Status Notes
BaseConfig 🟢 Stable Production-ready
Config Templates 🟢 Stable Production-ready
Environment Types 🟢 Stable Production-ready

Core Classes

BaseConfig

The main configuration class that provides environment variable support, type validation, and global configuration access.

Key Features:

  • Environment variable support
  • Type validation
  • Global configuration access
  • Nested configuration support

archipy.configs.base_config.R module-attribute 

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

archipy.configs.base_config.BaseConfig

Bases: BaseSettings

Base configuration class for ArchiPy applications.

This class provides a comprehensive configuration system that loads settings from multiple sources in the following priority order:

  1. pyproject.toml [tool.configs] section
  2. configs.toml or other specified TOML files
  3. Environment variables (.env file)
  4. OS-level environment variables
  5. Default class field values

The class implements the Singleton pattern via a global config instance that can be set once and accessed throughout the application.

Attributes:

Name Type Description
AUTH AuthConfig

Authentication and security settings
DATETIME DatetimeConfig

Date/time handling configuration
ELASTIC ElasticsearchConfig

Elasticsearch configuration
ELASTIC_APM ElasticsearchAPMConfig

Elasticsearch APM configuration
EMAIL EmailConfig

Email service configuration
ENVIRONMENT EnvironmentType

Application environment (dev, test, prod)
FASTAPI FastAPIConfig

FastAPI framework settings
FILE FileConfig

File handling configuration
GRPC GrpcConfig

gRPC service configuration
KAFKA KafkaConfig

Kafka integration configuration
KEYCLOAK KeycloakConfig

Keycloak integration configuration
MINIO MinioConfig

MinIO object storage configuration
PARSIAN_SHAPARAK ParsianShaparakConfig

Parsian Shaparak payment gateway configuration
POSTGRES_SQLALCHEMY PostgresSQLAlchemyConfig

PostgreSQL SQLAlchemy configuration
PROMETHEUS PrometheusConfig

Prometheus metrics configuration
REDIS RedisConfig

Redis cache configuration
SCYLLADB ScyllaDBConfig

ScyllaDB/Cassandra database configuration
SENTRY SentryConfig

Sentry error tracking configuration
SQLALCHEMY SQLAlchemyConfig

Database ORM configuration
SQLITE_SQLALCHEMY SqliteSQLAlchemyConfig

SQLite SQLAlchemy configuration
STARROCKS_SQLALCHEMY StarrocksSQLAlchemyConfig

Starrocks SQLAlchemy configuration
TEMPORAL TemporalConfig

Temporal workflow orchestration configuration

Examples:

>>> from archipy.configs.base_config import BaseConfig
>>>
>>> class MyAppConfig(BaseConfig):
...     # Override defaults
...     APP_NAME = "My Application"
...     DEBUG = True
...
...     # Custom configuration
...     FEATURE_FLAGS = {"new_ui": True, "advanced_search": False}
>>>
>>> # Set as global configuration
>>> config = MyAppConfig()
>>> BaseConfig.set_global(config)
>>>
>>> # Access from anywhere
>>> from archipy.configs.base_config import BaseConfig
>>> current_config = BaseConfig.global_config()
>>> app_name = current_config.APP_NAME  # "My Application"
Source code in archipy/configs/base_config.py 
 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
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
class BaseConfig[R](BaseSettings):
    """Base configuration class for ArchiPy applications.

    This class provides a comprehensive configuration system that loads settings
    from multiple sources in the following priority order:

    1. pyproject.toml [tool.configs] section
    2. configs.toml or other specified TOML files
    3. Environment variables (.env file)
    4. OS-level environment variables
    5. Default class field values

    The class implements the Singleton pattern via a global config instance that
    can be set once and accessed throughout the application.

    Attributes:
        AUTH (AuthConfig): Authentication and security settings
        DATETIME (DatetimeConfig): Date/time handling configuration
        ELASTIC (ElasticsearchConfig): Elasticsearch configuration
        ELASTIC_APM (ElasticsearchAPMConfig): Elasticsearch APM configuration
        EMAIL (EmailConfig): Email service configuration
        ENVIRONMENT (EnvironmentType): Application environment (dev, test, prod)
        FASTAPI (FastAPIConfig): FastAPI framework settings
        FILE (FileConfig): File handling configuration
        GRPC (GrpcConfig): gRPC service configuration
        KAFKA (KafkaConfig): Kafka integration configuration
        KEYCLOAK (KeycloakConfig): Keycloak integration configuration
        MINIO (MinioConfig): MinIO object storage configuration
        PARSIAN_SHAPARAK (ParsianShaparakConfig): Parsian Shaparak payment gateway configuration
        POSTGRES_SQLALCHEMY (PostgresSQLAlchemyConfig): PostgreSQL SQLAlchemy configuration
        PROMETHEUS (PrometheusConfig): Prometheus metrics configuration
        REDIS (RedisConfig): Redis cache configuration
        SCYLLADB (ScyllaDBConfig): ScyllaDB/Cassandra database configuration
        SENTRY (SentryConfig): Sentry error tracking configuration
        SQLALCHEMY (SQLAlchemyConfig): Database ORM configuration
        SQLITE_SQLALCHEMY (SqliteSQLAlchemyConfig): SQLite SQLAlchemy configuration
        STARROCKS_SQLALCHEMY (StarrocksSQLAlchemyConfig): Starrocks SQLAlchemy configuration
        TEMPORAL (TemporalConfig): Temporal workflow orchestration configuration

    Examples:
        >>> from archipy.configs.base_config import BaseConfig
        >>>
        >>> class MyAppConfig(BaseConfig):
        ...     # Override defaults
        ...     APP_NAME = "My Application"
        ...     DEBUG = True
        ...
        ...     # Custom configuration
        ...     FEATURE_FLAGS = {"new_ui": True, "advanced_search": False}
        >>>
        >>> # Set as global configuration
        >>> config = MyAppConfig()
        >>> BaseConfig.set_global(config)
        >>>
        >>> # Access from anywhere
        >>> from archipy.configs.base_config import BaseConfig
        >>> current_config = BaseConfig.global_config()
        >>> app_name = current_config.APP_NAME  # "My Application"
    """

    model_config = SettingsConfigDict(
        case_sensitive=True,
        pyproject_toml_depth=3,
        env_file=".env",
        pyproject_toml_table_header=("tool", "configs"),
        extra="ignore",
        env_nested_delimiter="__",
        env_ignore_empty=True,
    )

    __global_config: BaseConfig | None = None

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> tuple[PydanticBaseSettingsSource, ...]:
        """Customize the settings sources priority order.

        This method defines the priority order for configuration sources.

        Args:
            settings_cls: The settings class
            init_settings: Settings from initialization values
            env_settings: Settings from environment variables
            dotenv_settings: Settings from .env file
            file_secret_settings: Settings from secret files

        Returns:
            A tuple of configuration sources in priority order
        """
        return (
            file_secret_settings,
            PyprojectTomlConfigSettingsSource(settings_cls),
            TomlConfigSettingsSource(settings_cls),
            env_settings,
            dotenv_settings,
            init_settings,
        )

    AUTH: AuthConfig = AuthConfig()
    DATETIME: DatetimeConfig = DatetimeConfig()
    ELASTIC: ElasticsearchConfig = ElasticsearchConfig()
    ELASTIC_APM: ElasticsearchAPMConfig = ElasticsearchAPMConfig()
    EMAIL: EmailConfig = EmailConfig()
    ENVIRONMENT: EnvironmentType = EnvironmentType.LOCAL
    FASTAPI: FastAPIConfig = FastAPIConfig()
    FILE: FileConfig = FileConfig()
    GRPC: GrpcConfig = GrpcConfig()
    KAFKA: KafkaConfig = KafkaConfig()
    KEYCLOAK: KeycloakConfig = KeycloakConfig()
    MINIO: MinioConfig = MinioConfig()
    PARSIAN_SHAPARAK: ParsianShaparakConfig = ParsianShaparakConfig()
    PROMETHEUS: PrometheusConfig = PrometheusConfig()
    REDIS: RedisConfig = RedisConfig()
    SCYLLADB: ScyllaDBConfig = ScyllaDBConfig()
    SENTRY: SentryConfig = SentryConfig()
    SQLALCHEMY: SQLAlchemyConfig = SQLAlchemyConfig()
    STARROCKS_SQLALCHEMY: StarRocksSQLAlchemyConfig = StarRocksSQLAlchemyConfig()
    POSTGRES_SQLALCHEMY: PostgresSQLAlchemyConfig = PostgresSQLAlchemyConfig()
    SQLITE_SQLALCHEMY: SQLiteSQLAlchemyConfig = SQLiteSQLAlchemyConfig()
    TEMPORAL: TemporalConfig = TemporalConfig()
    LANGUAGE: LanguageType = LanguageType.FA

    def customize(self) -> None:
        """Customize configuration after loading.

        This method can be overridden in subclasses to perform
        custom configuration modifications after loading settings.
        """
        self.ELASTIC_APM.ENVIRONMENT = self.ENVIRONMENT

    @classmethod
    def global_config(cls) -> BaseConfig:
        """Retrieves the global configuration instance.

        Returns:
            BaseConfig: The global configuration instance.

        Raises:
            AssertionError: If the global config hasn't been set with
                BaseConfig.set_global()

        Examples:
            >>> config = BaseConfig.global_config()
            >>> redis_host = config.REDIS.MASTER_HOST
        """
        config_not_set_error = "You should set global configs with BaseConfig.set_global(MyConfig())"
        global_config = cls.__global_config
        if global_config is None:
            raise AssertionError(config_not_set_error)
        return global_config

    @classmethod
    def set_global(cls, config: BaseConfig) -> None:
        """Sets the global configuration instance.

        This method should be called once during application initialization
        to set the global configuration that will be used throughout the app.

        Args:
            config (BaseConfig): The configuration instance to use globally.

        Examples:
            >>> my_config = MyAppConfig(BaseConfig)
            >>> BaseConfig.set_global(my_config)
        """
        if hasattr(config, "customize") and callable(config.customize):
            config.customize()
        cls.__global_config = config

archipy.configs.base_config.BaseConfig.model_config class-attribute instance-attribute 

model_config = SettingsConfigDict(
    case_sensitive=True,
    pyproject_toml_depth=3,
    env_file=".env",
    pyproject_toml_table_header=("tool", "configs"),
    extra="ignore",
    env_nested_delimiter="__",
    env_ignore_empty=True,
)

archipy.configs.base_config.BaseConfig.AUTH class-attribute instance-attribute 

AUTH: AuthConfig = AuthConfig()

archipy.configs.base_config.BaseConfig.DATETIME class-attribute instance-attribute 

DATETIME: DatetimeConfig = DatetimeConfig()

archipy.configs.base_config.BaseConfig.ELASTIC class-attribute instance-attribute 

ELASTIC: ElasticsearchConfig = ElasticsearchConfig()

archipy.configs.base_config.BaseConfig.ELASTIC_APM class-attribute instance-attribute 

ELASTIC_APM: ElasticsearchAPMConfig = (
    ElasticsearchAPMConfig()
)

archipy.configs.base_config.BaseConfig.EMAIL class-attribute instance-attribute 

EMAIL: EmailConfig = EmailConfig()

archipy.configs.base_config.BaseConfig.ENVIRONMENT class-attribute instance-attribute 

ENVIRONMENT: EnvironmentType = LOCAL

archipy.configs.base_config.BaseConfig.FASTAPI class-attribute instance-attribute 

FASTAPI: FastAPIConfig = FastAPIConfig()

archipy.configs.base_config.BaseConfig.FILE class-attribute instance-attribute 

FILE: FileConfig = FileConfig()

archipy.configs.base_config.BaseConfig.GRPC class-attribute instance-attribute 

GRPC: GrpcConfig = GrpcConfig()

archipy.configs.base_config.BaseConfig.KAFKA class-attribute instance-attribute 

KAFKA: KafkaConfig = KafkaConfig()

archipy.configs.base_config.BaseConfig.KEYCLOAK class-attribute instance-attribute 

KEYCLOAK: KeycloakConfig = KeycloakConfig()

archipy.configs.base_config.BaseConfig.MINIO class-attribute instance-attribute 

MINIO: MinioConfig = MinioConfig()

archipy.configs.base_config.BaseConfig.PARSIAN_SHAPARAK class-attribute instance-attribute 

PARSIAN_SHAPARAK: ParsianShaparakConfig = (
    ParsianShaparakConfig()
)

archipy.configs.base_config.BaseConfig.PROMETHEUS class-attribute instance-attribute 

PROMETHEUS: PrometheusConfig = PrometheusConfig()

archipy.configs.base_config.BaseConfig.REDIS class-attribute instance-attribute 

REDIS: RedisConfig = RedisConfig()

archipy.configs.base_config.BaseConfig.SCYLLADB class-attribute instance-attribute 

SCYLLADB: ScyllaDBConfig = ScyllaDBConfig()

archipy.configs.base_config.BaseConfig.SENTRY class-attribute instance-attribute 

SENTRY: SentryConfig = SentryConfig()

archipy.configs.base_config.BaseConfig.SQLALCHEMY class-attribute instance-attribute 

SQLALCHEMY: SQLAlchemyConfig = SQLAlchemyConfig()

archipy.configs.base_config.BaseConfig.STARROCKS_SQLALCHEMY class-attribute instance-attribute 

STARROCKS_SQLALCHEMY: StarRocksSQLAlchemyConfig = (
    StarRocksSQLAlchemyConfig()
)

archipy.configs.base_config.BaseConfig.POSTGRES_SQLALCHEMY class-attribute instance-attribute 

POSTGRES_SQLALCHEMY: PostgresSQLAlchemyConfig = (
    PostgresSQLAlchemyConfig()
)

archipy.configs.base_config.BaseConfig.SQLITE_SQLALCHEMY class-attribute instance-attribute 

SQLITE_SQLALCHEMY: SQLiteSQLAlchemyConfig = (
    SQLiteSQLAlchemyConfig()
)

archipy.configs.base_config.BaseConfig.TEMPORAL class-attribute instance-attribute 

TEMPORAL: TemporalConfig = TemporalConfig()

archipy.configs.base_config.BaseConfig.LANGUAGE class-attribute instance-attribute 

LANGUAGE: LanguageType = FA

archipy.configs.base_config.BaseConfig.settings_customise_sources classmethod 

settings_customise_sources(
    settings_cls: type[BaseSettings],
    init_settings: PydanticBaseSettingsSource,
    env_settings: PydanticBaseSettingsSource,
    dotenv_settings: PydanticBaseSettingsSource,
    file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]

Customize the settings sources priority order.

This method defines the priority order for configuration sources.

Parameters:

Name Type Description Default
settings_cls type[BaseSettings]

The settings class

 required
init_settings PydanticBaseSettingsSource

Settings from initialization values

 required
env_settings PydanticBaseSettingsSource

Settings from environment variables

 required
dotenv_settings PydanticBaseSettingsSource

Settings from .env file

 required
file_secret_settings PydanticBaseSettingsSource

Settings from secret files

 required

Returns:

Type Description
tuple[PydanticBaseSettingsSource, ...]

A tuple of configuration sources in priority order
Source code in archipy/configs/base_config.py 
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
@classmethod
def settings_customise_sources(
    cls,
    settings_cls: type[BaseSettings],
    init_settings: PydanticBaseSettingsSource,
    env_settings: PydanticBaseSettingsSource,
    dotenv_settings: PydanticBaseSettingsSource,
    file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]:
    """Customize the settings sources priority order.

    This method defines the priority order for configuration sources.

    Args:
        settings_cls: The settings class
        init_settings: Settings from initialization values
        env_settings: Settings from environment variables
        dotenv_settings: Settings from .env file
        file_secret_settings: Settings from secret files

    Returns:
        A tuple of configuration sources in priority order
    """
    return (
        file_secret_settings,
        PyprojectTomlConfigSettingsSource(settings_cls),
        TomlConfigSettingsSource(settings_cls),
        env_settings,
        dotenv_settings,
        init_settings,
    )

archipy.configs.base_config.BaseConfig.customize 

customize() -> None

Customize configuration after loading.

This method can be overridden in subclasses to perform custom configuration modifications after loading settings.

Source code in archipy/configs/base_config.py 
177
178
179
180
181
182
183
def customize(self) -> None:
    """Customize configuration after loading.

    This method can be overridden in subclasses to perform
    custom configuration modifications after loading settings.
    """
    self.ELASTIC_APM.ENVIRONMENT = self.ENVIRONMENT

archipy.configs.base_config.BaseConfig.global_config classmethod 

global_config() -> BaseConfig

Retrieves the global configuration instance.

Returns:

Name Type Description
BaseConfig BaseConfig

The global configuration instance.

Raises:

Type Description
AssertionError

If the global config hasn't been set with BaseConfig.set_global()

Examples:

>>> config = BaseConfig.global_config()
>>> redis_host = config.REDIS.MASTER_HOST
Source code in archipy/configs/base_config.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@classmethod
def global_config(cls) -> BaseConfig:
    """Retrieves the global configuration instance.

    Returns:
        BaseConfig: The global configuration instance.

    Raises:
        AssertionError: If the global config hasn't been set with
            BaseConfig.set_global()

    Examples:
        >>> config = BaseConfig.global_config()
        >>> redis_host = config.REDIS.MASTER_HOST
    """
    config_not_set_error = "You should set global configs with BaseConfig.set_global(MyConfig())"
    global_config = cls.__global_config
    if global_config is None:
        raise AssertionError(config_not_set_error)
    return global_config

archipy.configs.base_config.BaseConfig.set_global classmethod 

set_global(config: BaseConfig) -> None

Sets the global configuration instance.

This method should be called once during application initialization to set the global configuration that will be used throughout the app.

Parameters:

Name Type Description Default
config BaseConfig

The configuration instance to use globally.

 required

Examples:

>>> my_config = MyAppConfig(BaseConfig)
>>> BaseConfig.set_global(my_config)
Source code in archipy/configs/base_config.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
@classmethod
def set_global(cls, config: BaseConfig) -> None:
    """Sets the global configuration instance.

    This method should be called once during application initialization
    to set the global configuration that will be used throughout the app.

    Args:
        config (BaseConfig): The configuration instance to use globally.

    Examples:
        >>> my_config = MyAppConfig(BaseConfig)
        >>> BaseConfig.set_global(my_config)
    """
    if hasattr(config, "customize") and callable(config.customize):
        config.customize()
    cls.__global_config = config

options: show_root_toc_entry: false heading_level: 3 members_order: alphabetical

Config Templates

For practical examples, see the Configuration Management Guide.

Database Configs

Bases: BaseModel

Configuration settings for SQLAlchemy ORM.

Controls database connection parameters, pooling behavior, and query execution settings.

Source code in archipy/configs/config_template.py 
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
class SQLAlchemyConfig(BaseModel):
    """Configuration settings for SQLAlchemy ORM.

    Controls database connection parameters, pooling behavior, and query execution settings.
    """

    DATABASE: str | None = Field(default=None, description="Database name")
    DRIVER_NAME: str = Field(default="postgresql+psycopg", description="Database driver name")
    ECHO: bool = Field(default=False, description="Whether to log SQL statements")
    ECHO_POOL: bool = Field(default=False, description="Whether to log connection pool events")
    ENABLE_FROM_LINTING: bool = Field(default=True, description="Whether to enable SQL linting")
    HIDE_PARAMETERS: bool = Field(default=False, description="Whether to hide SQL parameters in logs")
    HOST: str | None = Field(default=None, description="Database host")
    ISOLATION_LEVEL: str | None = Field(default="REPEATABLE READ", description="Transaction isolation level")
    PASSWORD: str | None = Field(default=None, description="Database password")
    POOL_MAX_OVERFLOW: int = Field(default=1, description="Maximum number of connections to allow in pool overflow")
    POOL_PRE_PING: bool = Field(default=True, description="Whether to ping connections before use")
    POOL_RECYCLE_SECONDS: int = Field(default=10 * 60, description="Number of seconds between connection recycling")
    POOL_RESET_ON_RETURN: str = Field(
        default="rollback",
        description="Action to take when returning connections to pool",
    )
    POOL_SIZE: int = Field(default=20, description="Number of connections to keep open in the pool")
    POOL_TIMEOUT: int = Field(default=30, description="Seconds to wait before giving up on getting a connection")
    POOL_USE_LIFO: bool = Field(default=True, description="Whether to use LIFO for connection pool")
    PORT: int | None = Field(default=5432, description="Database port")
    QUERY_CACHE_SIZE: int = Field(default=500, description="Size of the query cache")
    USERNAME: str | None = Field(default=None, description="Database username")

archipy.configs.config_template.SQLAlchemyConfig.DATABASE class-attribute instance-attribute 

DATABASE: str | None = Field(
    default=None, description="Database name"
)

archipy.configs.config_template.SQLAlchemyConfig.DRIVER_NAME class-attribute instance-attribute 

DRIVER_NAME: str = Field(
    default="postgresql+psycopg",
    description="Database driver name",
)

archipy.configs.config_template.SQLAlchemyConfig.ECHO class-attribute instance-attribute 

ECHO: bool = Field(
    default=False,
    description="Whether to log SQL statements",
)

archipy.configs.config_template.SQLAlchemyConfig.ECHO_POOL class-attribute instance-attribute 

ECHO_POOL: bool = Field(
    default=False,
    description="Whether to log connection pool events",
)

archipy.configs.config_template.SQLAlchemyConfig.ENABLE_FROM_LINTING class-attribute instance-attribute 

ENABLE_FROM_LINTING: bool = Field(
    default=True,
    description="Whether to enable SQL linting",
)

archipy.configs.config_template.SQLAlchemyConfig.HIDE_PARAMETERS class-attribute instance-attribute 

HIDE_PARAMETERS: bool = Field(
    default=False,
    description="Whether to hide SQL parameters in logs",
)

archipy.configs.config_template.SQLAlchemyConfig.HOST class-attribute instance-attribute 

HOST: str | None = Field(
    default=None, description="Database host"
)

archipy.configs.config_template.SQLAlchemyConfig.ISOLATION_LEVEL class-attribute instance-attribute 

ISOLATION_LEVEL: str | None = Field(
    default="REPEATABLE READ",
    description="Transaction isolation level",
)

archipy.configs.config_template.SQLAlchemyConfig.PASSWORD class-attribute instance-attribute 

PASSWORD: str | None = Field(
    default=None, description="Database password"
)

archipy.configs.config_template.SQLAlchemyConfig.POOL_MAX_OVERFLOW class-attribute instance-attribute 

POOL_MAX_OVERFLOW: int = Field(
    default=1,
    description="Maximum number of connections to allow in pool overflow",
)

archipy.configs.config_template.SQLAlchemyConfig.POOL_PRE_PING class-attribute instance-attribute 

POOL_PRE_PING: bool = Field(
    default=True,
    description="Whether to ping connections before use",
)

archipy.configs.config_template.SQLAlchemyConfig.POOL_RECYCLE_SECONDS class-attribute instance-attribute 

POOL_RECYCLE_SECONDS: int = Field(
    default=10 * 60,
    description="Number of seconds between connection recycling",
)

archipy.configs.config_template.SQLAlchemyConfig.POOL_RESET_ON_RETURN class-attribute instance-attribute 

POOL_RESET_ON_RETURN: str = Field(
    default="rollback",
    description="Action to take when returning connections to pool",
)

archipy.configs.config_template.SQLAlchemyConfig.POOL_SIZE class-attribute instance-attribute 

POOL_SIZE: int = Field(
    default=20,
    description="Number of connections to keep open in the pool",
)

archipy.configs.config_template.SQLAlchemyConfig.POOL_TIMEOUT class-attribute instance-attribute 

POOL_TIMEOUT: int = Field(
    default=30,
    description="Seconds to wait before giving up on getting a connection",
)

archipy.configs.config_template.SQLAlchemyConfig.POOL_USE_LIFO class-attribute instance-attribute 

POOL_USE_LIFO: bool = Field(
    default=True,
    description="Whether to use LIFO for connection pool",
)

archipy.configs.config_template.SQLAlchemyConfig.PORT class-attribute instance-attribute 

PORT: int | None = Field(
    default=5432, description="Database port"
)

archipy.configs.config_template.SQLAlchemyConfig.QUERY_CACHE_SIZE class-attribute instance-attribute 

QUERY_CACHE_SIZE: int = Field(
    default=500, description="Size of the query cache"
)

archipy.configs.config_template.SQLAlchemyConfig.USERNAME class-attribute instance-attribute 

USERNAME: str | None = Field(
    default=None, description="Database username"
)

options: show_root_toc_entry: false heading_level: 3

Bases: SQLAlchemyConfig

Configuration settings for PostgreSQL SQLAlchemy ORM.

Extends SQLAlchemyConfig with PostgreSQL-specific settings and URL building.

Source code in archipy/configs/config_template.py 
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
class PostgresSQLAlchemyConfig(SQLAlchemyConfig):
    """Configuration settings for PostgreSQL SQLAlchemy ORM.

    Extends SQLAlchemyConfig with PostgreSQL-specific settings and URL building.
    """

    POSTGRES_DSN: PostgresDsn | None = Field(default=None, description="PostgreSQL connection URL")

    @model_validator(mode="after")
    def build_connection_url(self) -> Self:
        """Build and populate DB_URL if not provided but all component parts are present.

        Returns:
            Self: The updated configuration instance.

        Raises:
            ValueError: If required connection parameters are missing.
        """
        if self.POSTGRES_DSN is not None:
            return self

        if all([self.USERNAME, self.HOST, self.PORT, self.DATABASE]):
            password_part = f":{self.PASSWORD}" if self.PASSWORD else ""
            self.POSTGRES_DSN = PostgresDsn(
                url=f"{self.DRIVER_NAME}://{self.USERNAME}{password_part}@{self.HOST}:{self.PORT}/{self.DATABASE}",
            )
        return self

    @model_validator(mode="after")
    def extract_connection_parts(self) -> Self:
        """Extract connection parts from DB_URL if provided but component parts are missing.

        Returns:
            Self: The updated configuration instance.

        Raises:
            ValueError: If the connection URL is invalid.
        """
        if self.POSTGRES_DSN is None:
            return self

        # Check if we need to extract components (if any are None)
        if any(x is None for x in [self.DRIVER_NAME, self.USERNAME, self.HOST, self.PORT, self.DATABASE]):
            url = str(self.POSTGRES_DSN)
            parsed = urlparse(url)

            # Extract scheme/driver (override default if URL scheme is different)
            if parsed.scheme and parsed.scheme != self.DRIVER_NAME:
                self.DRIVER_NAME = parsed.scheme

            # Extract username and password
            if parsed.netloc:
                auth_part = parsed.netloc.split("@")[0] if "@" in parsed.netloc else ""
                if ":" in auth_part:
                    username, password = auth_part.split(":", 1)
                    if self.USERNAME is None:
                        self.USERNAME = username
                    if self.PASSWORD is None:
                        self.PASSWORD = password
                elif auth_part and self.USERNAME is None:
                    self.USERNAME = auth_part

            # Extract host and port
            host_part = parsed.netloc.split("@")[-1] if "@" in parsed.netloc else parsed.netloc
            if ":" in host_part:
                host, port_str = host_part.split(":", 1)
                if self.HOST is None:
                    self.HOST = host
                if self.PORT is None:
                    with contextlib.suppress(ValueError):
                        self.PORT = int(port_str)
            elif host_part and self.HOST is None:
                self.HOST = host_part

            # Extract database name
            if self.DATABASE is None and parsed.path and parsed.path.startswith("/"):
                self.DATABASE = parsed.path[1:]

        return self

archipy.configs.config_template.PostgresSQLAlchemyConfig.POSTGRES_DSN class-attribute instance-attribute 

POSTGRES_DSN: PostgresDsn | None = Field(
    default=None, description="PostgreSQL connection URL"
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.DATABASE class-attribute instance-attribute 

DATABASE: str | None = Field(
    default=None, description="Database name"
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.DRIVER_NAME class-attribute instance-attribute 

DRIVER_NAME: str = Field(
    default="postgresql+psycopg",
    description="Database driver name",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.ECHO class-attribute instance-attribute 

ECHO: bool = Field(
    default=False,
    description="Whether to log SQL statements",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.ECHO_POOL class-attribute instance-attribute 

ECHO_POOL: bool = Field(
    default=False,
    description="Whether to log connection pool events",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.ENABLE_FROM_LINTING class-attribute instance-attribute 

ENABLE_FROM_LINTING: bool = Field(
    default=True,
    description="Whether to enable SQL linting",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.HIDE_PARAMETERS class-attribute instance-attribute 

HIDE_PARAMETERS: bool = Field(
    default=False,
    description="Whether to hide SQL parameters in logs",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.HOST class-attribute instance-attribute 

HOST: str | None = Field(
    default=None, description="Database host"
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.ISOLATION_LEVEL class-attribute instance-attribute 

ISOLATION_LEVEL: str | None = Field(
    default="REPEATABLE READ",
    description="Transaction isolation level",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.PASSWORD class-attribute instance-attribute 

PASSWORD: str | None = Field(
    default=None, description="Database password"
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.POOL_MAX_OVERFLOW class-attribute instance-attribute 

POOL_MAX_OVERFLOW: int = Field(
    default=1,
    description="Maximum number of connections to allow in pool overflow",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.POOL_PRE_PING class-attribute instance-attribute 

POOL_PRE_PING: bool = Field(
    default=True,
    description="Whether to ping connections before use",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.POOL_RECYCLE_SECONDS class-attribute instance-attribute 

POOL_RECYCLE_SECONDS: int = Field(
    default=10 * 60,
    description="Number of seconds between connection recycling",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.POOL_RESET_ON_RETURN class-attribute instance-attribute 

POOL_RESET_ON_RETURN: str = Field(
    default="rollback",
    description="Action to take when returning connections to pool",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.POOL_SIZE class-attribute instance-attribute 

POOL_SIZE: int = Field(
    default=20,
    description="Number of connections to keep open in the pool",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.POOL_TIMEOUT class-attribute instance-attribute 

POOL_TIMEOUT: int = Field(
    default=30,
    description="Seconds to wait before giving up on getting a connection",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.POOL_USE_LIFO class-attribute instance-attribute 

POOL_USE_LIFO: bool = Field(
    default=True,
    description="Whether to use LIFO for connection pool",
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.PORT class-attribute instance-attribute 

PORT: int | None = Field(
    default=5432, description="Database port"
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.QUERY_CACHE_SIZE class-attribute instance-attribute 

QUERY_CACHE_SIZE: int = Field(
    default=500, description="Size of the query cache"
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.USERNAME class-attribute instance-attribute 

USERNAME: str | None = Field(
    default=None, description="Database username"
)

archipy.configs.config_template.PostgresSQLAlchemyConfig.build_connection_url 

build_connection_url() -> Self

Build and populate DB_URL if not provided but all component parts are present.

Returns:

Name Type Description
Self Self

The updated configuration instance.

Raises:

Type Description
ValueError

If required connection parameters are missing.
Source code in archipy/configs/config_template.py 
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
@model_validator(mode="after")
def build_connection_url(self) -> Self:
    """Build and populate DB_URL if not provided but all component parts are present.

    Returns:
        Self: The updated configuration instance.

    Raises:
        ValueError: If required connection parameters are missing.
    """
    if self.POSTGRES_DSN is not None:
        return self

    if all([self.USERNAME, self.HOST, self.PORT, self.DATABASE]):
        password_part = f":{self.PASSWORD}" if self.PASSWORD else ""
        self.POSTGRES_DSN = PostgresDsn(
            url=f"{self.DRIVER_NAME}://{self.USERNAME}{password_part}@{self.HOST}:{self.PORT}/{self.DATABASE}",
        )
    return self

archipy.configs.config_template.PostgresSQLAlchemyConfig.extract_connection_parts 

extract_connection_parts() -> Self

Extract connection parts from DB_URL if provided but component parts are missing.

Returns:

Name Type Description
Self Self

The updated configuration instance.

Raises:

Type Description
ValueError

If the connection URL is invalid.
Source code in archipy/configs/config_template.py 
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
@model_validator(mode="after")
def extract_connection_parts(self) -> Self:
    """Extract connection parts from DB_URL if provided but component parts are missing.

    Returns:
        Self: The updated configuration instance.

    Raises:
        ValueError: If the connection URL is invalid.
    """
    if self.POSTGRES_DSN is None:
        return self

    # Check if we need to extract components (if any are None)
    if any(x is None for x in [self.DRIVER_NAME, self.USERNAME, self.HOST, self.PORT, self.DATABASE]):
        url = str(self.POSTGRES_DSN)
        parsed = urlparse(url)

        # Extract scheme/driver (override default if URL scheme is different)
        if parsed.scheme and parsed.scheme != self.DRIVER_NAME:
            self.DRIVER_NAME = parsed.scheme

        # Extract username and password
        if parsed.netloc:
            auth_part = parsed.netloc.split("@")[0] if "@" in parsed.netloc else ""
            if ":" in auth_part:
                username, password = auth_part.split(":", 1)
                if self.USERNAME is None:
                    self.USERNAME = username
                if self.PASSWORD is None:
                    self.PASSWORD = password
            elif auth_part and self.USERNAME is None:
                self.USERNAME = auth_part

        # Extract host and port
        host_part = parsed.netloc.split("@")[-1] if "@" in parsed.netloc else parsed.netloc
        if ":" in host_part:
            host, port_str = host_part.split(":", 1)
            if self.HOST is None:
                self.HOST = host
            if self.PORT is None:
                with contextlib.suppress(ValueError):
                    self.PORT = int(port_str)
        elif host_part and self.HOST is None:
            self.HOST = host_part

        # Extract database name
        if self.DATABASE is None and parsed.path and parsed.path.startswith("/"):
            self.DATABASE = parsed.path[1:]

    return self

options: show_root_toc_entry: false heading_level: 3

Bases: SQLAlchemyConfig

Configuration settings for SQLite SQLAlchemy ORM.

Extends SQLAlchemyConfig with SQLite-specific settings.

Source code in archipy/configs/config_template.py 
445
446
447
448
449
450
451
452
453
454
class SQLiteSQLAlchemyConfig(SQLAlchemyConfig):
    """Configuration settings for SQLite SQLAlchemy ORM.

    Extends SQLAlchemyConfig with SQLite-specific settings.
    """

    DRIVER_NAME: str = Field(default="sqlite+aiosqlite", description="SQLite driver name")
    DATABASE: str = Field(default=":memory:", description="SQLite database path")
    ISOLATION_LEVEL: str | None = Field(default=None, description="SQLite isolation level")
    PORT: int | None = Field(default=None, description="Not used for SQLite")

archipy.configs.config_template.SQLiteSQLAlchemyConfig.DRIVER_NAME class-attribute instance-attribute 

DRIVER_NAME: str = Field(
    default="sqlite+aiosqlite",
    description="SQLite driver name",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.DATABASE class-attribute instance-attribute 

DATABASE: str = Field(
    default=":memory:", description="SQLite database path"
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.ISOLATION_LEVEL class-attribute instance-attribute 

ISOLATION_LEVEL: str | None = Field(
    default=None, description="SQLite isolation level"
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.PORT class-attribute instance-attribute 

PORT: int | None = Field(
    default=None, description="Not used for SQLite"
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.ECHO class-attribute instance-attribute 

ECHO: bool = Field(
    default=False,
    description="Whether to log SQL statements",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.ECHO_POOL class-attribute instance-attribute 

ECHO_POOL: bool = Field(
    default=False,
    description="Whether to log connection pool events",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.ENABLE_FROM_LINTING class-attribute instance-attribute 

ENABLE_FROM_LINTING: bool = Field(
    default=True,
    description="Whether to enable SQL linting",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.HIDE_PARAMETERS class-attribute instance-attribute 

HIDE_PARAMETERS: bool = Field(
    default=False,
    description="Whether to hide SQL parameters in logs",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.HOST class-attribute instance-attribute 

HOST: str | None = Field(
    default=None, description="Database host"
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.PASSWORD class-attribute instance-attribute 

PASSWORD: str | None = Field(
    default=None, description="Database password"
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.POOL_MAX_OVERFLOW class-attribute instance-attribute 

POOL_MAX_OVERFLOW: int = Field(
    default=1,
    description="Maximum number of connections to allow in pool overflow",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.POOL_PRE_PING class-attribute instance-attribute 

POOL_PRE_PING: bool = Field(
    default=True,
    description="Whether to ping connections before use",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.POOL_RECYCLE_SECONDS class-attribute instance-attribute 

POOL_RECYCLE_SECONDS: int = Field(
    default=10 * 60,
    description="Number of seconds between connection recycling",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.POOL_RESET_ON_RETURN class-attribute instance-attribute 

POOL_RESET_ON_RETURN: str = Field(
    default="rollback",
    description="Action to take when returning connections to pool",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.POOL_SIZE class-attribute instance-attribute 

POOL_SIZE: int = Field(
    default=20,
    description="Number of connections to keep open in the pool",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.POOL_TIMEOUT class-attribute instance-attribute 

POOL_TIMEOUT: int = Field(
    default=30,
    description="Seconds to wait before giving up on getting a connection",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.POOL_USE_LIFO class-attribute instance-attribute 

POOL_USE_LIFO: bool = Field(
    default=True,
    description="Whether to use LIFO for connection pool",
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.QUERY_CACHE_SIZE class-attribute instance-attribute 

QUERY_CACHE_SIZE: int = Field(
    default=500, description="Size of the query cache"
)

archipy.configs.config_template.SQLiteSQLAlchemyConfig.USERNAME class-attribute instance-attribute 

USERNAME: str | None = Field(
    default=None, description="Database username"
)

options: show_root_toc_entry: false heading_level: 3

Bases: SQLAlchemyConfig

Configuration settings for Starrocks SQLAlchemy ORM.

Extends SQLAlchemyConfig with Starrocks-specific settings.

Note: StarRocks only supports READ COMMITTED isolation level. StarRocks uses MySQL protocol which requires explicit connection timeouts to prevent indefinite hangs on network issues.

Source code in archipy/configs/config_template.py 
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
class StarRocksSQLAlchemyConfig(SQLAlchemyConfig):
    """Configuration settings for Starrocks SQLAlchemy ORM.

    Extends SQLAlchemyConfig with Starrocks-specific settings.

    Note: StarRocks only supports READ COMMITTED isolation level.
    StarRocks uses MySQL protocol which requires explicit connection timeouts
    to prevent indefinite hangs on network issues.
    """

    DRIVER_NAME: str = Field(default="starrocks", description="StarRocks driver name")
    CATALOG: str | None = Field(default=None, description="Starrocks catalog name")
    ISOLATION_LEVEL: str = Field(
        default="READ COMMITTED",
        description="Transaction isolation level (StarRocks only supports READ COMMITTED)",
    )

    # Override timeout default for StarRocks (MySQL protocol requirement)
    CONNECT_TIMEOUT: int | None = Field(
        default=10,
        description="Timeout in seconds for establishing connection (MySQL protocol default: 10s)",
    )

    @field_validator("ISOLATION_LEVEL")
    @classmethod
    def validate_isolation_level(cls, v: str) -> str:
        """Validate that isolation level is READ COMMITTED for StarRocks.

        Args:
            v: The isolation level value to validate.

        Returns:
            The validated isolation level.

        Raises:
            ValueError: If the isolation level is not READ COMMITTED.
        """
        # Normalize the value (handle case variations and underscores)
        normalized = v.upper().replace("_", " ").strip()
        if normalized != "READ COMMITTED":
            raise ValueError(
                f"StarRocks only supports READ COMMITTED isolation level. Got: {v}. "
                "StarRocks does not support other isolation levels like REPEATABLE READ or SERIALIZABLE.",
            )
        return "READ COMMITTED"

archipy.configs.config_template.StarRocksSQLAlchemyConfig.DRIVER_NAME class-attribute instance-attribute 

DRIVER_NAME: str = Field(
    default="starrocks", description="StarRocks driver name"
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.CATALOG class-attribute instance-attribute 

CATALOG: str | None = Field(
    default=None, description="Starrocks catalog name"
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.ISOLATION_LEVEL class-attribute instance-attribute 

ISOLATION_LEVEL: str = Field(
    default="READ COMMITTED",
    description="Transaction isolation level (StarRocks only supports READ COMMITTED)",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.CONNECT_TIMEOUT class-attribute instance-attribute 

CONNECT_TIMEOUT: int | None = Field(
    default=10,
    description="Timeout in seconds for establishing connection (MySQL protocol default: 10s)",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.DATABASE class-attribute instance-attribute 

DATABASE: str | None = Field(
    default=None, description="Database name"
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.ECHO class-attribute instance-attribute 

ECHO: bool = Field(
    default=False,
    description="Whether to log SQL statements",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.ECHO_POOL class-attribute instance-attribute 

ECHO_POOL: bool = Field(
    default=False,
    description="Whether to log connection pool events",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.ENABLE_FROM_LINTING class-attribute instance-attribute 

ENABLE_FROM_LINTING: bool = Field(
    default=True,
    description="Whether to enable SQL linting",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.HIDE_PARAMETERS class-attribute instance-attribute 

HIDE_PARAMETERS: bool = Field(
    default=False,
    description="Whether to hide SQL parameters in logs",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.HOST class-attribute instance-attribute 

HOST: str | None = Field(
    default=None, description="Database host"
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.PASSWORD class-attribute instance-attribute 

PASSWORD: str | None = Field(
    default=None, description="Database password"
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.POOL_MAX_OVERFLOW class-attribute instance-attribute 

POOL_MAX_OVERFLOW: int = Field(
    default=1,
    description="Maximum number of connections to allow in pool overflow",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.POOL_PRE_PING class-attribute instance-attribute 

POOL_PRE_PING: bool = Field(
    default=True,
    description="Whether to ping connections before use",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.POOL_RECYCLE_SECONDS class-attribute instance-attribute 

POOL_RECYCLE_SECONDS: int = Field(
    default=10 * 60,
    description="Number of seconds between connection recycling",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.POOL_RESET_ON_RETURN class-attribute instance-attribute 

POOL_RESET_ON_RETURN: str = Field(
    default="rollback",
    description="Action to take when returning connections to pool",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.POOL_SIZE class-attribute instance-attribute 

POOL_SIZE: int = Field(
    default=20,
    description="Number of connections to keep open in the pool",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.POOL_TIMEOUT class-attribute instance-attribute 

POOL_TIMEOUT: int = Field(
    default=30,
    description="Seconds to wait before giving up on getting a connection",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.POOL_USE_LIFO class-attribute instance-attribute 

POOL_USE_LIFO: bool = Field(
    default=True,
    description="Whether to use LIFO for connection pool",
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.PORT class-attribute instance-attribute 

PORT: int | None = Field(
    default=5432, description="Database port"
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.QUERY_CACHE_SIZE class-attribute instance-attribute 

QUERY_CACHE_SIZE: int = Field(
    default=500, description="Size of the query cache"
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.USERNAME class-attribute instance-attribute 

USERNAME: str | None = Field(
    default=None, description="Database username"
)

archipy.configs.config_template.StarRocksSQLAlchemyConfig.validate_isolation_level classmethod 

validate_isolation_level(v: str) -> str

Validate that isolation level is READ COMMITTED for StarRocks.

Parameters:

Name Type Description Default
v str

The isolation level value to validate.

 required

Returns:

Type Description
str

The validated isolation level.

Raises:

Type Description
ValueError

If the isolation level is not READ COMMITTED.
Source code in archipy/configs/config_template.py 
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
@field_validator("ISOLATION_LEVEL")
@classmethod
def validate_isolation_level(cls, v: str) -> str:
    """Validate that isolation level is READ COMMITTED for StarRocks.

    Args:
        v: The isolation level value to validate.

    Returns:
        The validated isolation level.

    Raises:
        ValueError: If the isolation level is not READ COMMITTED.
    """
    # Normalize the value (handle case variations and underscores)
    normalized = v.upper().replace("_", " ").strip()
    if normalized != "READ COMMITTED":
        raise ValueError(
            f"StarRocks only supports READ COMMITTED isolation level. Got: {v}. "
            "StarRocks does not support other isolation levels like REPEATABLE READ or SERIALIZABLE.",
        )
    return "READ COMMITTED"

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for ScyllaDB/Cassandra connections and operations.

Contains settings related to ScyllaDB cluster connectivity, authentication, compression, consistency levels, connection management, retry policies, prepared statement caching, and health checks.

Source code in archipy/configs/config_template.py 
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
class ScyllaDBConfig(BaseModel):
    """Configuration settings for ScyllaDB/Cassandra connections and operations.

    Contains settings related to ScyllaDB cluster connectivity, authentication,
    compression, consistency levels, connection management, retry policies,
    prepared statement caching, and health checks.
    """

    CONTACT_POINTS: list[str] = Field(
        default=["127.0.0.1"],
        description="List of ScyllaDB node addresses for initial connection",
    )
    PORT: int = Field(
        default=9042,
        ge=1,
        le=65535,
        description="CQL native transport port number",
    )
    KEYSPACE: str | None = Field(
        default=None,
        description="Default keyspace name to use",
    )
    USERNAME: str | None = Field(
        default=None,
        description="Username for authentication",
    )
    PASSWORD: SecretStr | None = Field(
        default=None,
        description="Password for authentication",
    )
    PROTOCOL_VERSION: int = Field(
        default=4,
        ge=3,
        le=5,
        description="CQL protocol version (3-5)",
    )
    COMPRESSION: bool = Field(
        default=True,
        description="Enable LZ4 compression for network traffic",
    )
    CONNECT_TIMEOUT: int = Field(
        default=10,
        ge=1,
        description="Connection timeout in seconds",
    )
    REQUEST_TIMEOUT: int = Field(
        default=10,
        ge=1,
        description="Request timeout in seconds",
    )
    CONSISTENCY_LEVEL: Literal[
        "ONE",
        "TWO",
        "THREE",
        "QUORUM",
        "ALL",
        "LOCAL_QUORUM",
        "EACH_QUORUM",
        "LOCAL_ONE",
        "ANY",
    ] = Field(
        default="ONE",
        description="Default consistency level",
    )
    DISABLE_SHARD_AWARENESS: bool = Field(
        default=False,
        description="Disable shard awareness (useful for Docker/Testcontainer/NAT environments)",
    )
    ADDRESS_TRANSLATION_ENABLED: bool = Field(
        default=False,
        description="Enable address translation to redirect all discovered node connections to the first contact point. "
        "In Docker/Testcontainer/NAT environments, ScyllaDB nodes advertise their internal container IPs "
        "via gossip, which are unreachable from the host. When enabled, the driver translates all discovered "
        "addresses to the first configured contact point, allowing connections through Docker's port mapping.",
    )
    RETRY_POLICY: Literal["EXPONENTIAL_BACKOFF", "FALLTHROUGH"] = Field(
        default="EXPONENTIAL_BACKOFF",
        description="Retry policy type (uses native driver RetryPolicy). "
        "Options: 'EXPONENTIAL_BACKOFF' (retries with exponential backoff), "
        "'FALLTHROUGH' (never retries, propagates failures to application)",
    )
    RETRY_MAX_NUM_RETRIES: float = Field(
        default=3.0,
        ge=0.0,
        description="Maximum number of retries for ExponentialBackoffRetryPolicy",
    )
    RETRY_MIN_INTERVAL: float = Field(
        default=0.1,
        ge=0.0,
        description="Minimum interval in seconds between retries for ExponentialBackoffRetryPolicy",
    )
    RETRY_MAX_INTERVAL: float = Field(
        default=10.0,
        ge=0.0,
        description="Maximum interval in seconds between retries for ExponentialBackoffRetryPolicy",
    )
    ENABLE_PREPARED_STATEMENT_CACHE: bool = Field(
        default=True,
        description="Enable prepared statement caching",
    )
    PREPARED_STATEMENT_CACHE_SIZE: int = Field(
        default=100,
        ge=1,
        description="Maximum cached prepared statements",
    )
    PREPARED_STATEMENT_CACHE_TTL_SECONDS: int = Field(
        default=3600,
        ge=1,
        description="TTL for prepared statement cache in seconds (1 hour)",
    )
    HEALTH_CHECK_TIMEOUT: int = Field(
        default=5,
        ge=1,
        description="Timeout for health check queries in seconds",
    )
    ENABLE_CONNECTION_POOL_MONITORING: bool = Field(
        default=False,
        description="Enable connection pool monitoring and metrics",
    )

    # Connection Pool Configuration
    MAX_CONNECTIONS_PER_HOST: int = Field(
        default=2,
        ge=1,
        description="Maximum connections per host (recommended: 1-3 per CPU core)",
    )
    MIN_CONNECTIONS_PER_HOST: int = Field(
        default=1,
        ge=1,
        description="Minimum connections per host",
    )
    CORE_CONNECTIONS_PER_HOST: int = Field(
        default=1,
        ge=1,
        description="Core connections per host to maintain",
    )
    MAX_REQUESTS_PER_CONNECTION: int = Field(
        default=100,
        ge=1,
        description="Maximum concurrent requests per connection",
    )

    # Datacenter Configuration
    LOCAL_DC: str | None = Field(
        default=None,
        description="Local datacenter name for datacenter-aware routing",
    )
    REPLICATION_STRATEGY: Literal["SimpleStrategy", "NetworkTopologyStrategy"] = Field(
        default="SimpleStrategy",
        description="Replication strategy for keyspace creation",
    )
    REPLICATION_CONFIG: dict[str, int] | None = Field(
        default=None,
        description="Replication configuration (e.g., {'dc1': 3, 'dc2': 2} for NetworkTopologyStrategy)",
    )

    @model_validator(mode="after")
    def validate_contact_points(self) -> Self:
        """Validate that at least one contact point is provided."""
        if not self.CONTACT_POINTS or len(self.CONTACT_POINTS) == 0:
            raise InvalidArgumentError(
                argument_name="CONTACT_POINTS",
                additional_data={"error": "Empty contact points list"},
            )
        return self

    @model_validator(mode="after")
    def validate_authentication(self) -> Self:
        """Validate that both username and password are provided together."""
        if (self.USERNAME is None) != (self.PASSWORD is None):
            raise InvalidArgumentError(
                argument_name="authentication",
                additional_data={"error": "Both username and password must be provided together"},
            )
        return self

    @model_validator(mode="after")
    def validate_connection_pool(self) -> Self:
        """Validate connection pool configuration."""
        if self.MIN_CONNECTIONS_PER_HOST > self.MAX_CONNECTIONS_PER_HOST:
            raise InvalidArgumentError(
                argument_name="connection_pool",
                additional_data={"error": "MIN_CONNECTIONS_PER_HOST cannot exceed MAX_CONNECTIONS_PER_HOST"},
            )
        if self.CORE_CONNECTIONS_PER_HOST > self.MAX_CONNECTIONS_PER_HOST:
            raise InvalidArgumentError(
                argument_name="connection_pool",
                additional_data={"error": "CORE_CONNECTIONS_PER_HOST cannot exceed MAX_CONNECTIONS_PER_HOST"},
            )
        return self

    @model_validator(mode="after")
    def validate_replication_config(self) -> Self:
        """Validate replication configuration."""
        if self.REPLICATION_STRATEGY == "NetworkTopologyStrategy" and not self.REPLICATION_CONFIG:
            raise InvalidArgumentError(
                argument_name="replication_config",
                additional_data={"error": "REPLICATION_CONFIG required for NetworkTopologyStrategy"},
            )
        return self

archipy.configs.config_template.ScyllaDBConfig.CONTACT_POINTS class-attribute instance-attribute 

CONTACT_POINTS: list[str] = Field(
    default=["127.0.0.1"],
    description="List of ScyllaDB node addresses for initial connection",
)

archipy.configs.config_template.ScyllaDBConfig.PORT class-attribute instance-attribute 

PORT: int = Field(
    default=9042,
    ge=1,
    le=65535,
    description="CQL native transport port number",
)

archipy.configs.config_template.ScyllaDBConfig.KEYSPACE class-attribute instance-attribute 

KEYSPACE: str | None = Field(
    default=None, description="Default keyspace name to use"
)

archipy.configs.config_template.ScyllaDBConfig.USERNAME class-attribute instance-attribute 

USERNAME: str | None = Field(
    default=None, description="Username for authentication"
)

archipy.configs.config_template.ScyllaDBConfig.PASSWORD class-attribute instance-attribute 

PASSWORD: SecretStr | None = Field(
    default=None, description="Password for authentication"
)

archipy.configs.config_template.ScyllaDBConfig.PROTOCOL_VERSION class-attribute instance-attribute 

PROTOCOL_VERSION: int = Field(
    default=4,
    ge=3,
    le=5,
    description="CQL protocol version (3-5)",
)

archipy.configs.config_template.ScyllaDBConfig.COMPRESSION class-attribute instance-attribute 

COMPRESSION: bool = Field(
    default=True,
    description="Enable LZ4 compression for network traffic",
)

archipy.configs.config_template.ScyllaDBConfig.CONNECT_TIMEOUT class-attribute instance-attribute 

CONNECT_TIMEOUT: int = Field(
    default=10,
    ge=1,
    description="Connection timeout in seconds",
)

archipy.configs.config_template.ScyllaDBConfig.REQUEST_TIMEOUT class-attribute instance-attribute 

REQUEST_TIMEOUT: int = Field(
    default=10,
    ge=1,
    description="Request timeout in seconds",
)

archipy.configs.config_template.ScyllaDBConfig.CONSISTENCY_LEVEL class-attribute instance-attribute 

CONSISTENCY_LEVEL: Literal[
    "ONE",
    "TWO",
    "THREE",
    "QUORUM",
    "ALL",
    "LOCAL_QUORUM",
    "EACH_QUORUM",
    "LOCAL_ONE",
    "ANY",
] = Field(
    default="ONE", description="Default consistency level"
)

archipy.configs.config_template.ScyllaDBConfig.DISABLE_SHARD_AWARENESS class-attribute instance-attribute 

DISABLE_SHARD_AWARENESS: bool = Field(
    default=False,
    description="Disable shard awareness (useful for Docker/Testcontainer/NAT environments)",
)

archipy.configs.config_template.ScyllaDBConfig.ADDRESS_TRANSLATION_ENABLED class-attribute instance-attribute 

ADDRESS_TRANSLATION_ENABLED: bool = Field(
    default=False,
    description="Enable address translation to redirect all discovered node connections to the first contact point. In Docker/Testcontainer/NAT environments, ScyllaDB nodes advertise their internal container IPs via gossip, which are unreachable from the host. When enabled, the driver translates all discovered addresses to the first configured contact point, allowing connections through Docker's port mapping.",
)

archipy.configs.config_template.ScyllaDBConfig.RETRY_POLICY class-attribute instance-attribute 

RETRY_POLICY: Literal[
    "EXPONENTIAL_BACKOFF", "FALLTHROUGH"
] = Field(
    default="EXPONENTIAL_BACKOFF",
    description="Retry policy type (uses native driver RetryPolicy). Options: 'EXPONENTIAL_BACKOFF' (retries with exponential backoff), 'FALLTHROUGH' (never retries, propagates failures to application)",
)

archipy.configs.config_template.ScyllaDBConfig.RETRY_MAX_NUM_RETRIES class-attribute instance-attribute 

RETRY_MAX_NUM_RETRIES: float = Field(
    default=3.0,
    ge=0.0,
    description="Maximum number of retries for ExponentialBackoffRetryPolicy",
)

archipy.configs.config_template.ScyllaDBConfig.RETRY_MIN_INTERVAL class-attribute instance-attribute 

RETRY_MIN_INTERVAL: float = Field(
    default=0.1,
    ge=0.0,
    description="Minimum interval in seconds between retries for ExponentialBackoffRetryPolicy",
)

archipy.configs.config_template.ScyllaDBConfig.RETRY_MAX_INTERVAL class-attribute instance-attribute 

RETRY_MAX_INTERVAL: float = Field(
    default=10.0,
    ge=0.0,
    description="Maximum interval in seconds between retries for ExponentialBackoffRetryPolicy",
)

archipy.configs.config_template.ScyllaDBConfig.ENABLE_PREPARED_STATEMENT_CACHE class-attribute instance-attribute 

ENABLE_PREPARED_STATEMENT_CACHE: bool = Field(
    default=True,
    description="Enable prepared statement caching",
)

archipy.configs.config_template.ScyllaDBConfig.PREPARED_STATEMENT_CACHE_SIZE class-attribute instance-attribute 

PREPARED_STATEMENT_CACHE_SIZE: int = Field(
    default=100,
    ge=1,
    description="Maximum cached prepared statements",
)

archipy.configs.config_template.ScyllaDBConfig.PREPARED_STATEMENT_CACHE_TTL_SECONDS class-attribute instance-attribute 

PREPARED_STATEMENT_CACHE_TTL_SECONDS: int = Field(
    default=3600,
    ge=1,
    description="TTL for prepared statement cache in seconds (1 hour)",
)

archipy.configs.config_template.ScyllaDBConfig.HEALTH_CHECK_TIMEOUT class-attribute instance-attribute 

HEALTH_CHECK_TIMEOUT: int = Field(
    default=5,
    ge=1,
    description="Timeout for health check queries in seconds",
)

archipy.configs.config_template.ScyllaDBConfig.ENABLE_CONNECTION_POOL_MONITORING class-attribute instance-attribute 

ENABLE_CONNECTION_POOL_MONITORING: bool = Field(
    default=False,
    description="Enable connection pool monitoring and metrics",
)

archipy.configs.config_template.ScyllaDBConfig.MAX_CONNECTIONS_PER_HOST class-attribute instance-attribute 

MAX_CONNECTIONS_PER_HOST: int = Field(
    default=2,
    ge=1,
    description="Maximum connections per host (recommended: 1-3 per CPU core)",
)

archipy.configs.config_template.ScyllaDBConfig.MIN_CONNECTIONS_PER_HOST class-attribute instance-attribute 

MIN_CONNECTIONS_PER_HOST: int = Field(
    default=1,
    ge=1,
    description="Minimum connections per host",
)

archipy.configs.config_template.ScyllaDBConfig.CORE_CONNECTIONS_PER_HOST class-attribute instance-attribute 

CORE_CONNECTIONS_PER_HOST: int = Field(
    default=1,
    ge=1,
    description="Core connections per host to maintain",
)

archipy.configs.config_template.ScyllaDBConfig.MAX_REQUESTS_PER_CONNECTION class-attribute instance-attribute 

MAX_REQUESTS_PER_CONNECTION: int = Field(
    default=100,
    ge=1,
    description="Maximum concurrent requests per connection",
)

archipy.configs.config_template.ScyllaDBConfig.LOCAL_DC class-attribute instance-attribute 

LOCAL_DC: str | None = Field(
    default=None,
    description="Local datacenter name for datacenter-aware routing",
)

archipy.configs.config_template.ScyllaDBConfig.REPLICATION_STRATEGY class-attribute instance-attribute 

REPLICATION_STRATEGY: Literal[
    "SimpleStrategy", "NetworkTopologyStrategy"
] = Field(
    default="SimpleStrategy",
    description="Replication strategy for keyspace creation",
)

archipy.configs.config_template.ScyllaDBConfig.REPLICATION_CONFIG class-attribute instance-attribute 

REPLICATION_CONFIG: dict[str, int] | None = Field(
    default=None,
    description="Replication configuration (e.g., {'dc1': 3, 'dc2': 2} for NetworkTopologyStrategy)",
)

archipy.configs.config_template.ScyllaDBConfig.validate_contact_points 

validate_contact_points() -> Self

Validate that at least one contact point is provided.

Source code in archipy/configs/config_template.py 
1059
1060
1061
1062
1063
1064
1065
1066
1067
@model_validator(mode="after")
def validate_contact_points(self) -> Self:
    """Validate that at least one contact point is provided."""
    if not self.CONTACT_POINTS or len(self.CONTACT_POINTS) == 0:
        raise InvalidArgumentError(
            argument_name="CONTACT_POINTS",
            additional_data={"error": "Empty contact points list"},
        )
    return self

archipy.configs.config_template.ScyllaDBConfig.validate_authentication 

validate_authentication() -> Self

Validate that both username and password are provided together.

Source code in archipy/configs/config_template.py 
1069
1070
1071
1072
1073
1074
1075
1076
1077
@model_validator(mode="after")
def validate_authentication(self) -> Self:
    """Validate that both username and password are provided together."""
    if (self.USERNAME is None) != (self.PASSWORD is None):
        raise InvalidArgumentError(
            argument_name="authentication",
            additional_data={"error": "Both username and password must be provided together"},
        )
    return self

archipy.configs.config_template.ScyllaDBConfig.validate_connection_pool 

validate_connection_pool() -> Self

Validate connection pool configuration.

Source code in archipy/configs/config_template.py 
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
@model_validator(mode="after")
def validate_connection_pool(self) -> Self:
    """Validate connection pool configuration."""
    if self.MIN_CONNECTIONS_PER_HOST > self.MAX_CONNECTIONS_PER_HOST:
        raise InvalidArgumentError(
            argument_name="connection_pool",
            additional_data={"error": "MIN_CONNECTIONS_PER_HOST cannot exceed MAX_CONNECTIONS_PER_HOST"},
        )
    if self.CORE_CONNECTIONS_PER_HOST > self.MAX_CONNECTIONS_PER_HOST:
        raise InvalidArgumentError(
            argument_name="connection_pool",
            additional_data={"error": "CORE_CONNECTIONS_PER_HOST cannot exceed MAX_CONNECTIONS_PER_HOST"},
        )
    return self

archipy.configs.config_template.ScyllaDBConfig.validate_replication_config 

validate_replication_config() -> Self

Validate replication configuration.

Source code in archipy/configs/config_template.py 
1094
1095
1096
1097
1098
1099
1100
1101
1102
@model_validator(mode="after")
def validate_replication_config(self) -> Self:
    """Validate replication configuration."""
    if self.REPLICATION_STRATEGY == "NetworkTopologyStrategy" and not self.REPLICATION_CONFIG:
        raise InvalidArgumentError(
            argument_name="replication_config",
            additional_data={"error": "REPLICATION_CONFIG required for NetworkTopologyStrategy"},
        )
    return self

options: show_root_toc_entry: false heading_level: 3

Search & Analytics Configs

Bases: BaseModel

Configuration settings for Elasticsearch connections and operations.

Contains settings related to Elasticsearch server connectivity, authentication, TLS/SSL, request handling, node status management, and batch operation parameters.

Source code in archipy/configs/config_template.py 
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
80
81
82
83
84
85
86
87
88
89
90
91
class ElasticsearchConfig(BaseModel):
    """Configuration settings for Elasticsearch connections and operations.

    Contains settings related to Elasticsearch server connectivity, authentication,
    TLS/SSL, request handling, node status management, and batch operation parameters.
    """

    HOSTS: list[str] = Field(default=["https://localhost:9200"], description="List of Elasticsearch server hosts")
    HTTP_USER_NAME: str | None = None
    HTTP_PASSWORD: SecretStr | None = None
    API_KEY: str | None = None
    API_SECRET: SecretStr | None = None
    CA_CERTS: str | None = Field(default=None, description="Path to CA bundle for SSL verification")
    SSL_ASSERT_FINGERPRINT: str | None = Field(default=None, description="SSL certificate fingerprint for verification")
    VERIFY_CERTS: bool = Field(default=True, description="Whether to verify SSL certificates")
    CLIENT_CERT: str | None = Field(default=None, description="Path to client certificate for TLS authentication")
    CLIENT_KEY: str | None = Field(default=None, description="Path to client key for TLS authentication")
    HTTP_COMPRESS: bool = Field(default=True, description="Enable HTTP compression (gzip)")
    REQUEST_TIMEOUT: float | None = Field(default=1.0, description="Timeout for HTTP requests in seconds")
    MAX_RETRIES: int = Field(default=1, ge=0, description="Maximum number of retries per request")
    RETRY_ON_TIMEOUT: bool = Field(default=True, description="Retry on connection timeouts")
    RETRY_ON_STATUS: tuple[int, ...] = Field(default=(429, 502, 503, 504), description="HTTP status codes to retry on")
    IGNORE_STATUS: tuple[int, ...] = Field(default=(), description="HTTP status codes to ignore as errors")
    SNIFF_ON_START: bool = Field(default=False, description="Sniff nodes on client instantiation")
    SNIFF_BEFORE_REQUESTS: bool = Field(default=False, description="Sniff nodes before requests")
    SNIFF_ON_NODE_FAILURE: bool = Field(default=True, description="Sniff nodes on node failure")
    MIN_DELAY_BETWEEN_SNIFFING: float = Field(
        default=60.0,
        ge=0.0,
        description="Minimum delay between sniffing attempts in seconds",
    )
    NODE_SELECTOR_CLASS: str = Field(
        default="round_robin",
        description="Node selector strategy ('round_robin' or 'random')",
    )
    CONNECTIONS_PER_NODE: int = Field(default=10, ge=1, description="Number of HTTP connections per node")
    DEAD_NODE_BACKOFF_FACTOR: float = Field(
        default=1.0,
        ge=0.0,
        description="Factor for calculating node timeout duration after failures",
    )
    MAX_DEAD_NODE_BACKOFF: float = Field(
        default=300.0,
        ge=0.0,
        description="Maximum timeout duration for a dead node in seconds",
    )

    @model_validator(mode="after")
    def validate_tls_settings(self) -> Self:
        """Validate TLS-related settings to ensure compatibility."""
        if not self.VERIFY_CERTS and (self.CA_CERTS or self.SSL_ASSERT_FINGERPRINT):
            raise InvalidArgumentError()
        if self.CLIENT_CERT and not self.CLIENT_KEY:
            raise FailedPreconditionError()
        return self

    @model_validator(mode="after")
    def validate_sniffing_settings(self) -> Self:
        """Warn if sniffing is enabled with a load balancer."""
        if any([self.SNIFF_ON_START, self.SNIFF_BEFORE_REQUESTS, self.SNIFF_ON_NODE_FAILURE]):
            if len(self.HOSTS) == 1 and "localhost" not in self.HOSTS[0]:
                logger.warning("Warning: Sniffing may bypass load balancers or proxies, ensure this is intended.")
        return self

archipy.configs.config_template.ElasticsearchConfig.HOSTS class-attribute instance-attribute 

HOSTS: list[str] = Field(
    default=["https://localhost:9200"],
    description="List of Elasticsearch server hosts",
)

archipy.configs.config_template.ElasticsearchConfig.HTTP_USER_NAME class-attribute instance-attribute 

HTTP_USER_NAME: str | None = None

archipy.configs.config_template.ElasticsearchConfig.HTTP_PASSWORD class-attribute instance-attribute 

HTTP_PASSWORD: SecretStr | None = None

archipy.configs.config_template.ElasticsearchConfig.API_KEY class-attribute instance-attribute 

API_KEY: str | None = None

archipy.configs.config_template.ElasticsearchConfig.API_SECRET class-attribute instance-attribute 

API_SECRET: SecretStr | None = None

archipy.configs.config_template.ElasticsearchConfig.CA_CERTS class-attribute instance-attribute 

CA_CERTS: str | None = Field(
    default=None,
    description="Path to CA bundle for SSL verification",
)

archipy.configs.config_template.ElasticsearchConfig.SSL_ASSERT_FINGERPRINT class-attribute instance-attribute 

SSL_ASSERT_FINGERPRINT: str | None = Field(
    default=None,
    description="SSL certificate fingerprint for verification",
)

archipy.configs.config_template.ElasticsearchConfig.VERIFY_CERTS class-attribute instance-attribute 

VERIFY_CERTS: bool = Field(
    default=True,
    description="Whether to verify SSL certificates",
)

archipy.configs.config_template.ElasticsearchConfig.CLIENT_CERT class-attribute instance-attribute 

CLIENT_CERT: str | None = Field(
    default=None,
    description="Path to client certificate for TLS authentication",
)

archipy.configs.config_template.ElasticsearchConfig.CLIENT_KEY class-attribute instance-attribute 

CLIENT_KEY: str | None = Field(
    default=None,
    description="Path to client key for TLS authentication",
)

archipy.configs.config_template.ElasticsearchConfig.HTTP_COMPRESS class-attribute instance-attribute 

HTTP_COMPRESS: bool = Field(
    default=True,
    description="Enable HTTP compression (gzip)",
)

archipy.configs.config_template.ElasticsearchConfig.REQUEST_TIMEOUT class-attribute instance-attribute 

REQUEST_TIMEOUT: float | None = Field(
    default=1.0,
    description="Timeout for HTTP requests in seconds",
)

archipy.configs.config_template.ElasticsearchConfig.MAX_RETRIES class-attribute instance-attribute 

MAX_RETRIES: int = Field(
    default=1,
    ge=0,
    description="Maximum number of retries per request",
)

archipy.configs.config_template.ElasticsearchConfig.RETRY_ON_TIMEOUT class-attribute instance-attribute 

RETRY_ON_TIMEOUT: bool = Field(
    default=True, description="Retry on connection timeouts"
)

archipy.configs.config_template.ElasticsearchConfig.RETRY_ON_STATUS class-attribute instance-attribute 

RETRY_ON_STATUS: tuple[int, ...] = Field(
    default=(429, 502, 503, 504),
    description="HTTP status codes to retry on",
)

archipy.configs.config_template.ElasticsearchConfig.IGNORE_STATUS class-attribute instance-attribute 

IGNORE_STATUS: tuple[int, ...] = Field(
    default=(),
    description="HTTP status codes to ignore as errors",
)

archipy.configs.config_template.ElasticsearchConfig.SNIFF_ON_START class-attribute instance-attribute 

SNIFF_ON_START: bool = Field(
    default=False,
    description="Sniff nodes on client instantiation",
)

archipy.configs.config_template.ElasticsearchConfig.SNIFF_BEFORE_REQUESTS class-attribute instance-attribute 

SNIFF_BEFORE_REQUESTS: bool = Field(
    default=False, description="Sniff nodes before requests"
)

archipy.configs.config_template.ElasticsearchConfig.SNIFF_ON_NODE_FAILURE class-attribute instance-attribute 

SNIFF_ON_NODE_FAILURE: bool = Field(
    default=True, description="Sniff nodes on node failure"
)

archipy.configs.config_template.ElasticsearchConfig.MIN_DELAY_BETWEEN_SNIFFING class-attribute instance-attribute 

MIN_DELAY_BETWEEN_SNIFFING: float = Field(
    default=60.0,
    ge=0.0,
    description="Minimum delay between sniffing attempts in seconds",
)

archipy.configs.config_template.ElasticsearchConfig.NODE_SELECTOR_CLASS class-attribute instance-attribute 

NODE_SELECTOR_CLASS: str = Field(
    default="round_robin",
    description="Node selector strategy ('round_robin' or 'random')",
)

archipy.configs.config_template.ElasticsearchConfig.CONNECTIONS_PER_NODE class-attribute instance-attribute 

CONNECTIONS_PER_NODE: int = Field(
    default=10,
    ge=1,
    description="Number of HTTP connections per node",
)

archipy.configs.config_template.ElasticsearchConfig.DEAD_NODE_BACKOFF_FACTOR class-attribute instance-attribute 

DEAD_NODE_BACKOFF_FACTOR: float = Field(
    default=1.0,
    ge=0.0,
    description="Factor for calculating node timeout duration after failures",
)

archipy.configs.config_template.ElasticsearchConfig.MAX_DEAD_NODE_BACKOFF class-attribute instance-attribute 

MAX_DEAD_NODE_BACKOFF: float = Field(
    default=300.0,
    ge=0.0,
    description="Maximum timeout duration for a dead node in seconds",
)

archipy.configs.config_template.ElasticsearchConfig.validate_tls_settings 

validate_tls_settings() -> Self

Validate TLS-related settings to ensure compatibility.

Source code in archipy/configs/config_template.py 
76
77
78
79
80
81
82
83
@model_validator(mode="after")
def validate_tls_settings(self) -> Self:
    """Validate TLS-related settings to ensure compatibility."""
    if not self.VERIFY_CERTS and (self.CA_CERTS or self.SSL_ASSERT_FINGERPRINT):
        raise InvalidArgumentError()
    if self.CLIENT_CERT and not self.CLIENT_KEY:
        raise FailedPreconditionError()
    return self

archipy.configs.config_template.ElasticsearchConfig.validate_sniffing_settings 

validate_sniffing_settings() -> Self

Warn if sniffing is enabled with a load balancer.

Source code in archipy/configs/config_template.py 
85
86
87
88
89
90
91
@model_validator(mode="after")
def validate_sniffing_settings(self) -> Self:
    """Warn if sniffing is enabled with a load balancer."""
    if any([self.SNIFF_ON_START, self.SNIFF_BEFORE_REQUESTS, self.SNIFF_ON_NODE_FAILURE]):
        if len(self.HOSTS) == 1 and "localhost" not in self.HOSTS[0]:
            logger.warning("Warning: Sniffing may bypass load balancers or proxies, ensure this is intended.")
    return self

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for Elasticsearch APM (Application Performance Monitoring).

Controls behavior of the Elastic APM agent for application monitoring, tracing, and error reporting.

Source code in archipy/configs/config_template.py 
 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
class ElasticsearchAPMConfig(BaseModel):
    """Configuration settings for Elasticsearch APM (Application Performance Monitoring).

    Controls behavior of the Elastic APM agent for application monitoring, tracing,
    and error reporting.
    """

    API_REQUEST_SIZE: str = Field(default="768kb", description="Maximum size of API requests")
    API_REQUEST_TIME: str = Field(default="10s", description="Maximum time for API requests")
    AUTO_LOG_STACKS: bool = Field(default=True, description="Whether to automatically log stack traces")
    CAPTURE_BODY: str = Field(default="off", description="Level of request body capture")
    CAPTURE_HEADERS: bool = Field(default=False, description="Whether to capture HTTP headers")
    COLLECT_LOCAL_VARIABLES: str = Field(default="errors", description="Level of local variable collection")
    IS_ENABLED: bool = Field(default=False, description="Whether APM is enabled")
    ENVIRONMENT: str | None = Field(default=None, description="APM environment name")
    LOG_FILE: str = Field(default="", description="Path to APM log file")
    LOG_FILE_SIZE: str = Field(default="50mb", description="Maximum size of APM log file")
    RECORDING: bool = Field(default=True, description="Whether to record transactions")
    SECRET_TOKEN: str | None = Field(default=None, description="APM secret token")
    SERVER_TIMEOUT: str = Field(default="5s", description="Server timeout duration")
    SERVER_URL: str | None = Field(default=None, description="APM server URL")
    SERVICE_NAME: str = Field(default="unknown-python-service", description="Name of the service being monitored")
    SERVICE_VERSION: str | None = Field(default=None, description="Version of the service")
    TRANSACTION_SAMPLE_RATE: float = Field(default=0.001, description="Rate at which to sample transactions")
    API_KEY: str | None = Field(default=None, description="API key for authentication")

archipy.configs.config_template.ElasticsearchAPMConfig.API_REQUEST_SIZE class-attribute instance-attribute 

API_REQUEST_SIZE: str = Field(
    default="768kb",
    description="Maximum size of API requests",
)

archipy.configs.config_template.ElasticsearchAPMConfig.API_REQUEST_TIME class-attribute instance-attribute 

API_REQUEST_TIME: str = Field(
    default="10s",
    description="Maximum time for API requests",
)

archipy.configs.config_template.ElasticsearchAPMConfig.AUTO_LOG_STACKS class-attribute instance-attribute 

AUTO_LOG_STACKS: bool = Field(
    default=True,
    description="Whether to automatically log stack traces",
)

archipy.configs.config_template.ElasticsearchAPMConfig.CAPTURE_BODY class-attribute instance-attribute 

CAPTURE_BODY: str = Field(
    default="off",
    description="Level of request body capture",
)

archipy.configs.config_template.ElasticsearchAPMConfig.CAPTURE_HEADERS class-attribute instance-attribute 

CAPTURE_HEADERS: bool = Field(
    default=False,
    description="Whether to capture HTTP headers",
)

archipy.configs.config_template.ElasticsearchAPMConfig.COLLECT_LOCAL_VARIABLES class-attribute instance-attribute 

COLLECT_LOCAL_VARIABLES: str = Field(
    default="errors",
    description="Level of local variable collection",
)

archipy.configs.config_template.ElasticsearchAPMConfig.IS_ENABLED class-attribute instance-attribute 

IS_ENABLED: bool = Field(
    default=False, description="Whether APM is enabled"
)

archipy.configs.config_template.ElasticsearchAPMConfig.ENVIRONMENT class-attribute instance-attribute 

ENVIRONMENT: str | None = Field(
    default=None, description="APM environment name"
)

archipy.configs.config_template.ElasticsearchAPMConfig.LOG_FILE class-attribute instance-attribute 

LOG_FILE: str = Field(
    default="", description="Path to APM log file"
)

archipy.configs.config_template.ElasticsearchAPMConfig.LOG_FILE_SIZE class-attribute instance-attribute 

LOG_FILE_SIZE: str = Field(
    default="50mb",
    description="Maximum size of APM log file",
)

archipy.configs.config_template.ElasticsearchAPMConfig.RECORDING class-attribute instance-attribute 

RECORDING: bool = Field(
    default=True,
    description="Whether to record transactions",
)

archipy.configs.config_template.ElasticsearchAPMConfig.SECRET_TOKEN class-attribute instance-attribute 

SECRET_TOKEN: str | None = Field(
    default=None, description="APM secret token"
)

archipy.configs.config_template.ElasticsearchAPMConfig.SERVER_TIMEOUT class-attribute instance-attribute 

SERVER_TIMEOUT: str = Field(
    default="5s", description="Server timeout duration"
)

archipy.configs.config_template.ElasticsearchAPMConfig.SERVER_URL class-attribute instance-attribute 

SERVER_URL: str | None = Field(
    default=None, description="APM server URL"
)

archipy.configs.config_template.ElasticsearchAPMConfig.SERVICE_NAME class-attribute instance-attribute 

SERVICE_NAME: str = Field(
    default="unknown-python-service",
    description="Name of the service being monitored",
)

archipy.configs.config_template.ElasticsearchAPMConfig.SERVICE_VERSION class-attribute instance-attribute 

SERVICE_VERSION: str | None = Field(
    default=None, description="Version of the service"
)

archipy.configs.config_template.ElasticsearchAPMConfig.TRANSACTION_SAMPLE_RATE class-attribute instance-attribute 

TRANSACTION_SAMPLE_RATE: float = Field(
    default=0.001,
    description="Rate at which to sample transactions",
)

archipy.configs.config_template.ElasticsearchAPMConfig.API_KEY class-attribute instance-attribute 

API_KEY: str | None = Field(
    default=None, description="API key for authentication"
)

options: show_root_toc_entry: false heading_level: 3

Service Configs

Bases: BaseModel

Configuration settings for Redis cache integration.

Supports standalone, sentinel, and cluster deployments.

Source code in archipy/configs/config_template.py 
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
class RedisConfig(BaseModel):
    """Configuration settings for Redis cache integration.

    Supports standalone, sentinel, and cluster deployments.
    """

    # Deployment mode
    MODE: RedisMode = Field(default=RedisMode.STANDALONE, description="Redis deployment mode")

    # Standalone mode settings (existing)
    MASTER_HOST: str | None = Field(default="localhost", description="Redis master host (standalone/sentinel)")
    SLAVE_HOST: str | None = Field(default=None, description="Redis slave host (standalone)")

    # Cluster mode settings
    CLUSTER_NODES: list[str] = Field(default=[], description="List of cluster node addresses (host:port)")
    CLUSTER_REQUIRE_FULL_COVERAGE: bool = Field(default=True, description="Require full cluster coverage")
    CLUSTER_READ_FROM_REPLICAS: bool = Field(default=True, description="Allow reading from replica nodes")

    # Sentinel mode settings
    SENTINEL_NODES: list[str] = Field(default=[], description="List of sentinel addresses (host:port)")
    SENTINEL_SERVICE_NAME: str | None = Field(default=None, description="Master service name for sentinel")
    SENTINEL_SOCKET_TIMEOUT: float = Field(default=0.1, description="Sentinel socket timeout")

    # Common settings
    PORT: int = Field(default=6379, description="Default Redis server port")
    DATABASE: int = Field(default=0, description="Redis database number (not used in cluster)")
    PASSWORD: str | None = Field(default=None, description="Redis password")
    DECODE_RESPONSES: Literal[True] = Field(default=True, description="Whether to decode responses")
    VERSION: int = Field(default=7, description="Redis protocol version")
    HEALTH_CHECK_INTERVAL: int = Field(default=10, description="Health check interval in seconds")

    # Connection pooling
    MAX_CONNECTIONS: int = Field(default=50, description="Maximum connections per node")
    SOCKET_CONNECT_TIMEOUT: float = Field(default=5.0, description="Socket connection timeout")
    SOCKET_TIMEOUT: float = Field(default=5.0, description="Socket operation timeout")

    @model_validator(mode="after")
    def validate_mode_configuration(self) -> Self:
        """Validate mode-specific configuration."""
        if self.MODE == RedisMode.CLUSTER:
            if not self.CLUSTER_NODES:
                raise ValueError("CLUSTER_NODES must be provided when MODE is 'cluster'")
            if self.DATABASE != 0:
                logger.warning("DATABASE setting ignored in cluster mode")

        elif self.MODE == RedisMode.SENTINEL:
            if not self.SENTINEL_NODES or not self.SENTINEL_SERVICE_NAME:
                raise ValueError("SENTINEL_NODES and SENTINEL_SERVICE_NAME required for sentinel mode")

        elif self.MODE == RedisMode.STANDALONE:
            if not self.MASTER_HOST:
                raise ValueError("MASTER_HOST required for standalone mode")

        return self

archipy.configs.config_template.RedisConfig.MODE class-attribute instance-attribute 

MODE: RedisMode = Field(
    default=STANDALONE, description="Redis deployment mode"
)

archipy.configs.config_template.RedisConfig.MASTER_HOST class-attribute instance-attribute 

MASTER_HOST: str | None = Field(
    default="localhost",
    description="Redis master host (standalone/sentinel)",
)

archipy.configs.config_template.RedisConfig.SLAVE_HOST class-attribute instance-attribute 

SLAVE_HOST: str | None = Field(
    default=None,
    description="Redis slave host (standalone)",
)

archipy.configs.config_template.RedisConfig.CLUSTER_NODES class-attribute instance-attribute 

CLUSTER_NODES: list[str] = Field(
    default=[],
    description="List of cluster node addresses (host:port)",
)

archipy.configs.config_template.RedisConfig.CLUSTER_REQUIRE_FULL_COVERAGE class-attribute instance-attribute 

CLUSTER_REQUIRE_FULL_COVERAGE: bool = Field(
    default=True,
    description="Require full cluster coverage",
)

archipy.configs.config_template.RedisConfig.CLUSTER_READ_FROM_REPLICAS class-attribute instance-attribute 

CLUSTER_READ_FROM_REPLICAS: bool = Field(
    default=True,
    description="Allow reading from replica nodes",
)

archipy.configs.config_template.RedisConfig.SENTINEL_NODES class-attribute instance-attribute 

SENTINEL_NODES: list[str] = Field(
    default=[],
    description="List of sentinel addresses (host:port)",
)

archipy.configs.config_template.RedisConfig.SENTINEL_SERVICE_NAME class-attribute instance-attribute 

SENTINEL_SERVICE_NAME: str | None = Field(
    default=None,
    description="Master service name for sentinel",
)

archipy.configs.config_template.RedisConfig.SENTINEL_SOCKET_TIMEOUT class-attribute instance-attribute 

SENTINEL_SOCKET_TIMEOUT: float = Field(
    default=0.1, description="Sentinel socket timeout"
)

archipy.configs.config_template.RedisConfig.PORT class-attribute instance-attribute 

PORT: int = Field(
    default=6379, description="Default Redis server port"
)

archipy.configs.config_template.RedisConfig.DATABASE class-attribute instance-attribute 

DATABASE: int = Field(
    default=0,
    description="Redis database number (not used in cluster)",
)

archipy.configs.config_template.RedisConfig.PASSWORD class-attribute instance-attribute 

PASSWORD: str | None = Field(
    default=None, description="Redis password"
)

archipy.configs.config_template.RedisConfig.DECODE_RESPONSES class-attribute instance-attribute 

DECODE_RESPONSES: Literal[True] = Field(
    default=True, description="Whether to decode responses"
)

archipy.configs.config_template.RedisConfig.VERSION class-attribute instance-attribute 

VERSION: int = Field(
    default=7, description="Redis protocol version"
)

archipy.configs.config_template.RedisConfig.HEALTH_CHECK_INTERVAL class-attribute instance-attribute 

HEALTH_CHECK_INTERVAL: int = Field(
    default=10,
    description="Health check interval in seconds",
)

archipy.configs.config_template.RedisConfig.MAX_CONNECTIONS class-attribute instance-attribute 

MAX_CONNECTIONS: int = Field(
    default=50, description="Maximum connections per node"
)

archipy.configs.config_template.RedisConfig.SOCKET_CONNECT_TIMEOUT class-attribute instance-attribute 

SOCKET_CONNECT_TIMEOUT: float = Field(
    default=5.0, description="Socket connection timeout"
)

archipy.configs.config_template.RedisConfig.SOCKET_TIMEOUT class-attribute instance-attribute 

SOCKET_TIMEOUT: float = Field(
    default=5.0, description="Socket operation timeout"
)

archipy.configs.config_template.RedisConfig.validate_mode_configuration 

validate_mode_configuration() -> Self

Validate mode-specific configuration.

Source code in archipy/configs/config_template.py 
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
@model_validator(mode="after")
def validate_mode_configuration(self) -> Self:
    """Validate mode-specific configuration."""
    if self.MODE == RedisMode.CLUSTER:
        if not self.CLUSTER_NODES:
            raise ValueError("CLUSTER_NODES must be provided when MODE is 'cluster'")
        if self.DATABASE != 0:
            logger.warning("DATABASE setting ignored in cluster mode")

    elif self.MODE == RedisMode.SENTINEL:
        if not self.SENTINEL_NODES or not self.SENTINEL_SERVICE_NAME:
            raise ValueError("SENTINEL_NODES and SENTINEL_SERVICE_NAME required for sentinel mode")

    elif self.MODE == RedisMode.STANDALONE:
        if not self.MASTER_HOST:
            raise ValueError("MASTER_HOST required for standalone mode")

    return self

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for Apache Kafka integration.

Controls Kafka producer and consumer behavior, including broker connections, message delivery guarantees, and performance settings.

Source code in archipy/configs/config_template.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
class KafkaConfig(BaseModel):
    """Configuration settings for Apache Kafka integration.

    Controls Kafka producer and consumer behavior, including broker connections,
    message delivery guarantees, and performance settings.
    """

    BROKERS_LIST: list[str] = Field(default=["localhost:9092"], description="List of Kafka broker addresses")
    SECURITY_PROTOCOL: str = Field(default="PLAINTEXT", description="Security protocol for Kafka connections")
    SASL_MECHANISM: str | None = Field(default=None, description="SASL mechanism for authentication")
    USERNAME: str | None = Field(default=None, description="Username for SASL authentication")
    PASSWORD: SecretStr | None = Field(default=None, description="Password for SASL authentication")
    SSL_CA_FILE: str | None = Field(default=None, description="Path to SSL CA certificate file")
    SSL_CERT_FILE: str | None = Field(default=None, description="Path to SSL certificate file")
    SSL_KEY_FILE: str | None = Field(default=None, description="Path to SSL key file")
    ACKS: Literal["0", "1", "all"] = Field(default="all", description="Acknowledgment mode for producers")
    AUTO_OFFSET_RESET: Literal["earliest", "latest", "none"] = Field(
        default="earliest",
        description="Offset reset policy for consumers",
    )
    ENABLE_AUTO_COMMIT: bool = Field(default=False, description="Enable auto-commit for consumer offsets")
    FETCH_MIN_BYTES: int = Field(default=1, ge=1, description="Minimum bytes to fetch per poll")
    SESSION_TIMEOUT_MS: int = Field(default=10000, ge=1000, description="Consumer session timeout (ms)")
    HEARTBEAT_INTERVAL_MS: int = Field(default=3000, ge=100, description="Consumer heartbeat interval (ms)")
    REQUEST_TIMEOUT_MS: int = Field(default=30000, ge=1000, description="Request timeout (ms)")
    DELIVERY_TIMEOUT_MS: int = Field(default=120000, ge=1000, description="Message delivery timeout (ms)")
    COMPRESSION_TYPE: Literal["none", "gzip", "snappy", "lz4", "zstd"] | None = Field(
        default=None,
        description="Compression type for messages",
    )
    LINGER_MS: int = Field(default=0, ge=0, description="Time to buffer messages before sending (ms)")
    BATCH_SIZE: int = Field(default=16384, ge=0, description="Maximum batch size in bytes")
    MAX_IN_FLIGHT_REQUESTS: int = Field(default=5, ge=1, description="Maximum unacknowledged requests per connection")
    RETRIES: int = Field(default=5, ge=0, description="Number of retries for failed producer requests")
    LIST_TOPICS_TIMEOUT_MS: int = Field(default=5000, ge=1000, description="Timeout for listing topics (ms)")
    CLIENT_ID: str = Field(default="kafka-client", description="Client identifier")
    CONNECTIONS_MAX_IDLE_MS: int = Field(
        default=540000,
        description="Close idle connections after this number of milliseconds",
    )
    ENABLE_IDEMPOTENCE: bool = Field(default=False, description="Enable idempotent producer for exactly-once delivery")
    TRANSACTIONAL_ID: str | None = Field(default=None, description="Transactional ID for the producer")
    ISOLATION_LEVEL: Literal["read_uncommitted", "read_committed"] = Field(
        default="read_uncommitted",
        description="Isolation level for consumer",
    )
    MAX_POLL_INTERVAL_MS: int = Field(default=300000, ge=1000, description="Maximum time between poll invocations")
    PARTITION_ASSIGNMENT_STRATEGY: str = Field(
        default="range",
        description="Partition assignment strategy for consumer",
    )
    FETCH_MAX_BYTES: int = Field(
        default=52428800,
        ge=0,
        description="Maximum amount of data the server returns for a fetch request",
    )
    MAX_PARTITION_FETCH_BYTES: int = Field(
        default=1048576,
        ge=0,
        description="Maximum amount of data per partition the server returns",
    )
    QUEUE_BUFFERING_MAX_MESSAGES: int = Field(
        default=100000,
        ge=0,
        description="Maximum number of messages allowed on the producer queue",
    )
    STATISTICS_INTERVAL_MS: int = Field(
        default=0,
        ge=0,
        description="Frequency in milliseconds to send statistics data",
    )

    # Async adapter settings (AIOProducer / AIOConsumer)
    CONSUMER_MAX_WORKERS: int = Field(default=2, ge=1, description="Thread pool workers for AIOConsumer")
    PRODUCER_BATCH_SIZE: int = Field(default=1000, ge=1, description="Max messages per AIOProducer batch")
    PRODUCER_BUFFER_TIMEOUT: float = Field(
        default=1.0,
        ge=0.0,
        description="Buffer flush timeout for AIOProducer (seconds)",
    )
    PRODUCER_MAX_WORKERS: int = Field(default=4, ge=1, description="Thread pool workers for AIOProducer")

    @model_validator(mode="after")
    def validate_security_settings(self) -> KafkaConfig:
        """Validate security-related settings for Kafka configuration.

        Ensures that SASL authentication settings are properly configured when
        using SASL security protocols, and warns about missing SSL certificates
        when SSL is enabled.

        Returns:
            KafkaConfig: The validated configuration instance.

        Raises:
            ValueError: If SASL authentication is incomplete.
        """
        if self.SECURITY_PROTOCOL in ["SASL_PLAINTEXT", "SASL_SSL"]:
            if not (self.SASL_MECHANISM and self.USERNAME and self.PASSWORD):
                raise ValueError("SASL authentication requires SASL_MECHANISM, USERNAME, and PASSWORD to be set.")
        if self.SECURITY_PROTOCOL == "SSL":
            if not (self.SSL_CA_FILE or self.SSL_CERT_FILE or self.SSL_KEY_FILE):
                logger.warning("SSL enabled but no SSL certificates provided; this may cause connection issues.")
        return self

    @model_validator(mode="after")
    def validate_consumer_settings(self) -> KafkaConfig:
        """Validate consumer-specific settings for Kafka configuration.

        Ensures that auto-commit and offset reset settings are compatible,
        and that heartbeat interval is less than session timeout.

        Returns:
            KafkaConfig: The validated configuration instance.

        Raises:
            ValueError: If consumer settings are incompatible.
        """
        if self.ENABLE_AUTO_COMMIT and self.AUTO_OFFSET_RESET == "none":
            raise ValueError("ENABLE_AUTO_COMMIT cannot be True when AUTO_OFFSET_RESET is 'none'.")
        if self.HEARTBEAT_INTERVAL_MS >= self.SESSION_TIMEOUT_MS:
            raise ValueError("HEARTBEAT_INTERVAL_MS must be less than SESSION_TIMEOUT_MS.")
        return self

    @model_validator(mode="after")
    def validate_idempotence_and_transactions(self) -> KafkaConfig:
        """Validate idempotence and transaction settings for Kafka configuration.

        Ensures that idempotence is properly configured with 'all' acknowledgments,
        and that transactional producers have idempotence enabled.

        Returns:
            KafkaConfig: The validated configuration instance.

        Raises:
            ValueError: If idempotence or transaction settings are invalid.
        """
        if self.ENABLE_IDEMPOTENCE and self.ACKS != "all":
            raise ValueError("ENABLE_IDEMPOTENCE requires ACKS to be 'all'.")
        if self.TRANSACTIONAL_ID is not None and not self.ENABLE_IDEMPOTENCE:
            raise ValueError("TRANSACTIONAL_ID requires ENABLE_IDEMPOTENCE to be True.")
        return self

archipy.configs.config_template.KafkaConfig.BROKERS_LIST class-attribute instance-attribute 

BROKERS_LIST: list[str] = Field(
    default=["localhost:9092"],
    description="List of Kafka broker addresses",
)

archipy.configs.config_template.KafkaConfig.SECURITY_PROTOCOL class-attribute instance-attribute 

SECURITY_PROTOCOL: str = Field(
    default="PLAINTEXT",
    description="Security protocol for Kafka connections",
)

archipy.configs.config_template.KafkaConfig.SASL_MECHANISM class-attribute instance-attribute 

SASL_MECHANISM: str | None = Field(
    default=None,
    description="SASL mechanism for authentication",
)

archipy.configs.config_template.KafkaConfig.USERNAME class-attribute instance-attribute 

USERNAME: str | None = Field(
    default=None,
    description="Username for SASL authentication",
)

archipy.configs.config_template.KafkaConfig.PASSWORD class-attribute instance-attribute 

PASSWORD: SecretStr | None = Field(
    default=None,
    description="Password for SASL authentication",
)

archipy.configs.config_template.KafkaConfig.SSL_CA_FILE class-attribute instance-attribute 

SSL_CA_FILE: str | None = Field(
    default=None,
    description="Path to SSL CA certificate file",
)

archipy.configs.config_template.KafkaConfig.SSL_CERT_FILE class-attribute instance-attribute 

SSL_CERT_FILE: str | None = Field(
    default=None, description="Path to SSL certificate file"
)

archipy.configs.config_template.KafkaConfig.SSL_KEY_FILE class-attribute instance-attribute 

SSL_KEY_FILE: str | None = Field(
    default=None, description="Path to SSL key file"
)

archipy.configs.config_template.KafkaConfig.ACKS class-attribute instance-attribute 

ACKS: Literal["0", "1", "all"] = Field(
    default="all",
    description="Acknowledgment mode for producers",
)

archipy.configs.config_template.KafkaConfig.AUTO_OFFSET_RESET class-attribute instance-attribute 

AUTO_OFFSET_RESET: Literal["earliest", "latest", "none"] = (
    Field(
        default="earliest",
        description="Offset reset policy for consumers",
    )
)

archipy.configs.config_template.KafkaConfig.ENABLE_AUTO_COMMIT class-attribute instance-attribute 

ENABLE_AUTO_COMMIT: bool = Field(
    default=False,
    description="Enable auto-commit for consumer offsets",
)

archipy.configs.config_template.KafkaConfig.FETCH_MIN_BYTES class-attribute instance-attribute 

FETCH_MIN_BYTES: int = Field(
    default=1,
    ge=1,
    description="Minimum bytes to fetch per poll",
)

archipy.configs.config_template.KafkaConfig.SESSION_TIMEOUT_MS class-attribute instance-attribute 

SESSION_TIMEOUT_MS: int = Field(
    default=10000,
    ge=1000,
    description="Consumer session timeout (ms)",
)

archipy.configs.config_template.KafkaConfig.HEARTBEAT_INTERVAL_MS class-attribute instance-attribute 

HEARTBEAT_INTERVAL_MS: int = Field(
    default=3000,
    ge=100,
    description="Consumer heartbeat interval (ms)",
)

archipy.configs.config_template.KafkaConfig.REQUEST_TIMEOUT_MS class-attribute instance-attribute 

REQUEST_TIMEOUT_MS: int = Field(
    default=30000,
    ge=1000,
    description="Request timeout (ms)",
)

archipy.configs.config_template.KafkaConfig.DELIVERY_TIMEOUT_MS class-attribute instance-attribute 

DELIVERY_TIMEOUT_MS: int = Field(
    default=120000,
    ge=1000,
    description="Message delivery timeout (ms)",
)

archipy.configs.config_template.KafkaConfig.COMPRESSION_TYPE class-attribute instance-attribute 

COMPRESSION_TYPE: (
    Literal["none", "gzip", "snappy", "lz4", "zstd"] | None
) = Field(
    default=None,
    description="Compression type for messages",
)

archipy.configs.config_template.KafkaConfig.LINGER_MS class-attribute instance-attribute 

LINGER_MS: int = Field(
    default=0,
    ge=0,
    description="Time to buffer messages before sending (ms)",
)

archipy.configs.config_template.KafkaConfig.BATCH_SIZE class-attribute instance-attribute 

BATCH_SIZE: int = Field(
    default=16384,
    ge=0,
    description="Maximum batch size in bytes",
)

archipy.configs.config_template.KafkaConfig.MAX_IN_FLIGHT_REQUESTS class-attribute instance-attribute 

MAX_IN_FLIGHT_REQUESTS: int = Field(
    default=5,
    ge=1,
    description="Maximum unacknowledged requests per connection",
)

archipy.configs.config_template.KafkaConfig.RETRIES class-attribute instance-attribute 

RETRIES: int = Field(
    default=5,
    ge=0,
    description="Number of retries for failed producer requests",
)

archipy.configs.config_template.KafkaConfig.LIST_TOPICS_TIMEOUT_MS class-attribute instance-attribute 

LIST_TOPICS_TIMEOUT_MS: int = Field(
    default=5000,
    ge=1000,
    description="Timeout for listing topics (ms)",
)

archipy.configs.config_template.KafkaConfig.CLIENT_ID class-attribute instance-attribute 

CLIENT_ID: str = Field(
    default="kafka-client", description="Client identifier"
)

archipy.configs.config_template.KafkaConfig.CONNECTIONS_MAX_IDLE_MS class-attribute instance-attribute 

CONNECTIONS_MAX_IDLE_MS: int = Field(
    default=540000,
    description="Close idle connections after this number of milliseconds",
)

archipy.configs.config_template.KafkaConfig.ENABLE_IDEMPOTENCE class-attribute instance-attribute 

ENABLE_IDEMPOTENCE: bool = Field(
    default=False,
    description="Enable idempotent producer for exactly-once delivery",
)

archipy.configs.config_template.KafkaConfig.TRANSACTIONAL_ID class-attribute instance-attribute 

TRANSACTIONAL_ID: str | None = Field(
    default=None,
    description="Transactional ID for the producer",
)

archipy.configs.config_template.KafkaConfig.ISOLATION_LEVEL class-attribute instance-attribute 

ISOLATION_LEVEL: Literal[
    "read_uncommitted", "read_committed"
] = Field(
    default="read_uncommitted",
    description="Isolation level for consumer",
)

archipy.configs.config_template.KafkaConfig.MAX_POLL_INTERVAL_MS class-attribute instance-attribute 

MAX_POLL_INTERVAL_MS: int = Field(
    default=300000,
    ge=1000,
    description="Maximum time between poll invocations",
)

archipy.configs.config_template.KafkaConfig.PARTITION_ASSIGNMENT_STRATEGY class-attribute instance-attribute 

PARTITION_ASSIGNMENT_STRATEGY: str = Field(
    default="range",
    description="Partition assignment strategy for consumer",
)

archipy.configs.config_template.KafkaConfig.FETCH_MAX_BYTES class-attribute instance-attribute 

FETCH_MAX_BYTES: int = Field(
    default=52428800,
    ge=0,
    description="Maximum amount of data the server returns for a fetch request",
)

archipy.configs.config_template.KafkaConfig.MAX_PARTITION_FETCH_BYTES class-attribute instance-attribute 

MAX_PARTITION_FETCH_BYTES: int = Field(
    default=1048576,
    ge=0,
    description="Maximum amount of data per partition the server returns",
)

archipy.configs.config_template.KafkaConfig.QUEUE_BUFFERING_MAX_MESSAGES class-attribute instance-attribute 

QUEUE_BUFFERING_MAX_MESSAGES: int = Field(
    default=100000,
    ge=0,
    description="Maximum number of messages allowed on the producer queue",
)

archipy.configs.config_template.KafkaConfig.STATISTICS_INTERVAL_MS class-attribute instance-attribute 

STATISTICS_INTERVAL_MS: int = Field(
    default=0,
    ge=0,
    description="Frequency in milliseconds to send statistics data",
)

archipy.configs.config_template.KafkaConfig.CONSUMER_MAX_WORKERS class-attribute instance-attribute 

CONSUMER_MAX_WORKERS: int = Field(
    default=2,
    ge=1,
    description="Thread pool workers for AIOConsumer",
)

archipy.configs.config_template.KafkaConfig.PRODUCER_BATCH_SIZE class-attribute instance-attribute 

PRODUCER_BATCH_SIZE: int = Field(
    default=1000,
    ge=1,
    description="Max messages per AIOProducer batch",
)

archipy.configs.config_template.KafkaConfig.PRODUCER_BUFFER_TIMEOUT class-attribute instance-attribute 

PRODUCER_BUFFER_TIMEOUT: float = Field(
    default=1.0,
    ge=0.0,
    description="Buffer flush timeout for AIOProducer (seconds)",
)

archipy.configs.config_template.KafkaConfig.PRODUCER_MAX_WORKERS class-attribute instance-attribute 

PRODUCER_MAX_WORKERS: int = Field(
    default=4,
    ge=1,
    description="Thread pool workers for AIOProducer",
)

archipy.configs.config_template.KafkaConfig.validate_security_settings 

validate_security_settings() -> KafkaConfig

Validate security-related settings for Kafka configuration.

Ensures that SASL authentication settings are properly configured when using SASL security protocols, and warns about missing SSL certificates when SSL is enabled.

Returns:

Name Type Description
KafkaConfig KafkaConfig

The validated configuration instance.

Raises:

Type Description
ValueError

If SASL authentication is incomplete.
Source code in archipy/configs/config_template.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
@model_validator(mode="after")
def validate_security_settings(self) -> KafkaConfig:
    """Validate security-related settings for Kafka configuration.

    Ensures that SASL authentication settings are properly configured when
    using SASL security protocols, and warns about missing SSL certificates
    when SSL is enabled.

    Returns:
        KafkaConfig: The validated configuration instance.

    Raises:
        ValueError: If SASL authentication is incomplete.
    """
    if self.SECURITY_PROTOCOL in ["SASL_PLAINTEXT", "SASL_SSL"]:
        if not (self.SASL_MECHANISM and self.USERNAME and self.PASSWORD):
            raise ValueError("SASL authentication requires SASL_MECHANISM, USERNAME, and PASSWORD to be set.")
    if self.SECURITY_PROTOCOL == "SSL":
        if not (self.SSL_CA_FILE or self.SSL_CERT_FILE or self.SSL_KEY_FILE):
            logger.warning("SSL enabled but no SSL certificates provided; this may cause connection issues.")
    return self

archipy.configs.config_template.KafkaConfig.validate_consumer_settings 

validate_consumer_settings() -> KafkaConfig

Validate consumer-specific settings for Kafka configuration.

Ensures that auto-commit and offset reset settings are compatible, and that heartbeat interval is less than session timeout.

Returns:

Name Type Description
KafkaConfig KafkaConfig

The validated configuration instance.

Raises:

Type Description
ValueError

If consumer settings are incompatible.
Source code in archipy/configs/config_template.py 
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
@model_validator(mode="after")
def validate_consumer_settings(self) -> KafkaConfig:
    """Validate consumer-specific settings for Kafka configuration.

    Ensures that auto-commit and offset reset settings are compatible,
    and that heartbeat interval is less than session timeout.

    Returns:
        KafkaConfig: The validated configuration instance.

    Raises:
        ValueError: If consumer settings are incompatible.
    """
    if self.ENABLE_AUTO_COMMIT and self.AUTO_OFFSET_RESET == "none":
        raise ValueError("ENABLE_AUTO_COMMIT cannot be True when AUTO_OFFSET_RESET is 'none'.")
    if self.HEARTBEAT_INTERVAL_MS >= self.SESSION_TIMEOUT_MS:
        raise ValueError("HEARTBEAT_INTERVAL_MS must be less than SESSION_TIMEOUT_MS.")
    return self

archipy.configs.config_template.KafkaConfig.validate_idempotence_and_transactions 

validate_idempotence_and_transactions() -> KafkaConfig

Validate idempotence and transaction settings for Kafka configuration.

Ensures that idempotence is properly configured with 'all' acknowledgments, and that transactional producers have idempotence enabled.

Returns:

Name Type Description
KafkaConfig KafkaConfig

The validated configuration instance.

Raises:

Type Description
ValueError

If idempotence or transaction settings are invalid.
Source code in archipy/configs/config_template.py 
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
@model_validator(mode="after")
def validate_idempotence_and_transactions(self) -> KafkaConfig:
    """Validate idempotence and transaction settings for Kafka configuration.

    Ensures that idempotence is properly configured with 'all' acknowledgments,
    and that transactional producers have idempotence enabled.

    Returns:
        KafkaConfig: The validated configuration instance.

    Raises:
        ValueError: If idempotence or transaction settings are invalid.
    """
    if self.ENABLE_IDEMPOTENCE and self.ACKS != "all":
        raise ValueError("ENABLE_IDEMPOTENCE requires ACKS to be 'all'.")
    if self.TRANSACTIONAL_ID is not None and not self.ENABLE_IDEMPOTENCE:
        raise ValueError("TRANSACTIONAL_ID requires ENABLE_IDEMPOTENCE to be True.")
    return self

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for email service integration.

Controls SMTP server connection parameters, authentication, and email sending behavior.

Source code in archipy/configs/config_template.py 
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
class EmailConfig(BaseModel):
    """Configuration settings for email service integration.

    Controls SMTP server connection parameters, authentication,
    and email sending behavior.
    """

    SMTP_SERVER: str | None = Field(default=None, description="SMTP server host")
    SMTP_PORT: int = Field(default=587, description="SMTP server port")
    USERNAME: str | None = Field(default=None, description="SMTP username")
    PASSWORD: str | None = Field(default=None, description="SMTP password")
    POOL_SIZE: int = Field(default=5, description="Connection pool size")
    CONNECTION_TIMEOUT: int = Field(default=30, description="Connection timeout in seconds")
    MAX_RETRIES: int = Field(default=3, description="Maximum retry attempts")
    ATTACHMENT_MAX_SIZE: int = Field(default=5 * 1024 * 1024, description="Maximum attachment size in bytes")

archipy.configs.config_template.EmailConfig.SMTP_SERVER class-attribute instance-attribute 

SMTP_SERVER: str | None = Field(
    default=None, description="SMTP server host"
)

archipy.configs.config_template.EmailConfig.SMTP_PORT class-attribute instance-attribute 

SMTP_PORT: int = Field(
    default=587, description="SMTP server port"
)

archipy.configs.config_template.EmailConfig.USERNAME class-attribute instance-attribute 

USERNAME: str | None = Field(
    default=None, description="SMTP username"
)

archipy.configs.config_template.EmailConfig.PASSWORD class-attribute instance-attribute 

PASSWORD: str | None = Field(
    default=None, description="SMTP password"
)

archipy.configs.config_template.EmailConfig.POOL_SIZE class-attribute instance-attribute 

POOL_SIZE: int = Field(
    default=5, description="Connection pool size"
)

archipy.configs.config_template.EmailConfig.CONNECTION_TIMEOUT class-attribute instance-attribute 

CONNECTION_TIMEOUT: int = Field(
    default=30, description="Connection timeout in seconds"
)

archipy.configs.config_template.EmailConfig.MAX_RETRIES class-attribute instance-attribute 

MAX_RETRIES: int = Field(
    default=3, description="Maximum retry attempts"
)

archipy.configs.config_template.EmailConfig.ATTACHMENT_MAX_SIZE class-attribute instance-attribute 

ATTACHMENT_MAX_SIZE: int = Field(
    default=5 * 1024 * 1024,
    description="Maximum attachment size in bytes",
)

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for MinIO/S3 object storage integration.

Controls connection parameters and authentication for S3-compatible object storage services using boto3.

Source code in archipy/configs/config_template.py 
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
class MinioConfig(BaseModel):
    """Configuration settings for MinIO/S3 object storage integration.

    Controls connection parameters and authentication for S3-compatible
    object storage services using boto3.
    """

    ENDPOINT: str | None = Field(default=None, description="MinIO/S3 server endpoint")
    ACCESS_KEY: str | None = Field(default=None, description="Access key for authentication")
    SECRET_KEY: str | None = Field(default=None, description="Secret key for authentication")
    SECURE: bool = Field(default=False, description="Whether to use secure (HTTPS) connection")
    SESSION_TOKEN: str | None = Field(default=None, description="Session token for temporary credentials")
    REGION: str | None = Field(default=None, description="AWS region for S3 compatibility")
    ADDRESSING_STYLE: Literal["auto", "path", "virtual"] = Field(
        default="auto",
        description="S3 addressing style for URLs",
    )
    SIGNATURE_VERSION: str = Field(default="s3v4", description="AWS signature version (s3v4 recommended)")
    CONNECT_TIMEOUT: int = Field(default=60, description="Connection timeout in seconds")
    READ_TIMEOUT: int = Field(default=60, description="Read timeout in seconds")
    MAX_POOL_CONNECTIONS: int = Field(default=10, description="Maximum number of connections in the pool")
    RETRIES_MAX_ATTEMPTS: int = Field(default=3, description="Maximum retry attempts for failed requests")
    RETRIES_MODE: Literal["legacy", "standard", "adaptive"] = Field(default="standard", description="Retry mode")
    USE_SSL: bool | None = Field(default=None, description="Explicitly set SSL usage (overrides SECURE if set)")
    VERIFY_SSL: bool = Field(default=True, description="Verify SSL certificates")

archipy.configs.config_template.MinioConfig.ENDPOINT class-attribute instance-attribute 

ENDPOINT: str | None = Field(
    default=None, description="MinIO/S3 server endpoint"
)

archipy.configs.config_template.MinioConfig.ACCESS_KEY class-attribute instance-attribute 

ACCESS_KEY: str | None = Field(
    default=None,
    description="Access key for authentication",
)

archipy.configs.config_template.MinioConfig.SECRET_KEY class-attribute instance-attribute 

SECRET_KEY: str | None = Field(
    default=None,
    description="Secret key for authentication",
)

archipy.configs.config_template.MinioConfig.SECURE class-attribute instance-attribute 

SECURE: bool = Field(
    default=False,
    description="Whether to use secure (HTTPS) connection",
)

archipy.configs.config_template.MinioConfig.SESSION_TOKEN class-attribute instance-attribute 

SESSION_TOKEN: str | None = Field(
    default=None,
    description="Session token for temporary credentials",
)

archipy.configs.config_template.MinioConfig.REGION class-attribute instance-attribute 

REGION: str | None = Field(
    default=None,
    description="AWS region for S3 compatibility",
)

archipy.configs.config_template.MinioConfig.ADDRESSING_STYLE class-attribute instance-attribute 

ADDRESSING_STYLE: Literal["auto", "path", "virtual"] = (
    Field(
        default="auto",
        description="S3 addressing style for URLs",
    )
)

archipy.configs.config_template.MinioConfig.SIGNATURE_VERSION class-attribute instance-attribute 

SIGNATURE_VERSION: str = Field(
    default="s3v4",
    description="AWS signature version (s3v4 recommended)",
)

archipy.configs.config_template.MinioConfig.CONNECT_TIMEOUT class-attribute instance-attribute 

CONNECT_TIMEOUT: int = Field(
    default=60, description="Connection timeout in seconds"
)

archipy.configs.config_template.MinioConfig.READ_TIMEOUT class-attribute instance-attribute 

READ_TIMEOUT: int = Field(
    default=60, description="Read timeout in seconds"
)

archipy.configs.config_template.MinioConfig.MAX_POOL_CONNECTIONS class-attribute instance-attribute 

MAX_POOL_CONNECTIONS: int = Field(
    default=10,
    description="Maximum number of connections in the pool",
)

archipy.configs.config_template.MinioConfig.RETRIES_MAX_ATTEMPTS class-attribute instance-attribute 

RETRIES_MAX_ATTEMPTS: int = Field(
    default=3,
    description="Maximum retry attempts for failed requests",
)

archipy.configs.config_template.MinioConfig.RETRIES_MODE class-attribute instance-attribute 

RETRIES_MODE: Literal["legacy", "standard", "adaptive"] = (
    Field(default="standard", description="Retry mode")
)

archipy.configs.config_template.MinioConfig.USE_SSL class-attribute instance-attribute 

USE_SSL: bool | None = Field(
    default=None,
    description="Explicitly set SSL usage (overrides SECURE if set)",
)

archipy.configs.config_template.MinioConfig.VERIFY_SSL class-attribute instance-attribute 

VERIFY_SSL: bool = Field(
    default=True, description="Verify SSL certificates"
)

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for Keycloak integration.

Controls connection parameters and authentication settings for the Keycloak identity and access management service.

Source code in archipy/configs/config_template.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
class KeycloakConfig(BaseModel):
    """Configuration settings for Keycloak integration.

    Controls connection parameters and authentication settings for the Keycloak
    identity and access management service.
    """

    SERVER_URL: str | None = None
    CLIENT_ID: str | None = None
    REALM_NAME: str = "master"
    CLIENT_SECRET_KEY: str | None = None
    VERIFY_SSL: bool = True
    TIMEOUT: int = 10
    IS_ADMIN_MODE_ENABLED: bool = False
    ADMIN_USERNAME: str | None = None
    ADMIN_PASSWORD: str | None = None
    ADMIN_REALM_NAME: str = "master"

archipy.configs.config_template.KeycloakConfig.SERVER_URL class-attribute instance-attribute 

SERVER_URL: str | None = None

archipy.configs.config_template.KeycloakConfig.CLIENT_ID class-attribute instance-attribute 

CLIENT_ID: str | None = None

archipy.configs.config_template.KeycloakConfig.REALM_NAME class-attribute instance-attribute 

REALM_NAME: str = 'master'

archipy.configs.config_template.KeycloakConfig.CLIENT_SECRET_KEY class-attribute instance-attribute 

CLIENT_SECRET_KEY: str | None = None

archipy.configs.config_template.KeycloakConfig.VERIFY_SSL class-attribute instance-attribute 

VERIFY_SSL: bool = True

archipy.configs.config_template.KeycloakConfig.TIMEOUT class-attribute instance-attribute 

TIMEOUT: int = 10

archipy.configs.config_template.KeycloakConfig.IS_ADMIN_MODE_ENABLED class-attribute instance-attribute 

IS_ADMIN_MODE_ENABLED: bool = False

archipy.configs.config_template.KeycloakConfig.ADMIN_USERNAME class-attribute instance-attribute 

ADMIN_USERNAME: str | None = None

archipy.configs.config_template.KeycloakConfig.ADMIN_PASSWORD class-attribute instance-attribute 

ADMIN_PASSWORD: str | None = None

archipy.configs.config_template.KeycloakConfig.ADMIN_REALM_NAME class-attribute instance-attribute 

ADMIN_REALM_NAME: str = 'master'

options: show_root_toc_entry: false heading_level: 3

Web Framework Configs

Bases: BaseModel

Configuration settings for FastAPI applications.

Controls FastAPI application behavior, including server settings, middleware, documentation, and performance parameters.

Source code in archipy/configs/config_template.py 
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
class FastAPIConfig(BaseModel):
    """Configuration settings for FastAPI applications.

    Controls FastAPI application behavior, including server settings, middleware,
    documentation, and performance parameters.
    """

    PROJECT_NAME: str = Field(default="project_name", description="Name of the FastAPI project")
    API_PREFIX: str = Field(default="/api", description="URL prefix for API endpoints")

    ACCESS_LOG: bool = Field(default=True, description="Whether to enable access logging")
    BACKLOG: int = Field(default=2048, description="Maximum number of queued connections")
    DATE_HEADER: bool = Field(default=True, description="Whether to include date header in responses")
    FORWARDED_ALLOW_IPS: list[str] | None = Field(default=None, description="List of allowed forwarded IPs")
    LIMIT_CONCURRENCY: int | None = Field(default=None, description="Maximum concurrent requests")
    LIMIT_MAX_REQUESTS: int | None = Field(default=None, description="Maximum number of requests")
    CORS_MIDDLEWARE_ALLOW_CREDENTIALS: bool = Field(default=True, description="Whether to allow credentials in CORS")
    CORS_MIDDLEWARE_ALLOW_HEADERS: list[str] = Field(default=["*"], description="Allowed CORS headers")
    CORS_MIDDLEWARE_ALLOW_METHODS: list[str] = Field(default=["*"], description="Allowed CORS methods")
    CORS_MIDDLEWARE_ALLOW_ORIGINS: list[str] = Field(default=["*"], description="Allowed CORS origins")
    PROXY_HEADERS: bool = Field(default=True, description="Whether to trust proxy headers")
    RELOAD: bool = Field(default=False, description="Whether to enable auto-reload")
    SERVER_HEADER: bool = Field(default=True, description="Whether to include server header")
    SERVE_HOST: str = Field(
        default="0.0.0.0",
        description="Host to serve the application on",
    )  # Deliberate binding to all interfaces for containerized deployments
    SERVE_PORT: int = Field(default=8100, description="Port to serve the application on")
    TIMEOUT_GRACEFUL_SHUTDOWN: int | None = Field(default=None, description="Graceful shutdown timeout")
    TIMEOUT_KEEP_ALIVE: int = Field(default=5, description="Keep-alive timeout")
    WORKERS_COUNT: int = Field(default=4, description="Number of worker processes")
    WS_MAX_SIZE: int = Field(default=16777216, description="Maximum WebSocket message size")
    WS_PER_MESSAGE_DEFLATE: bool = Field(default=True, description="Whether to enable WebSocket compression")
    WS_PING_INTERVAL: float = Field(default=20.0, description="WebSocket ping interval")
    WS_PING_TIMEOUT: float = Field(default=20.0, description="WebSocket ping timeout")
    OPENAPI_URL: str | None = Field(default=None, description="URL for OpenAPI schema")
    DOCS_URL: str | None = Field(default=None, description="URL for API documentation")
    RE_DOC_URL: str | None = Field(default=None, description="URL for ReDoc documentation")
    SWAGGER_UI_PARAMS: dict[str, str] | None = Field(
        default={"docExpansion": "none"},
        description="Swagger UI parameters",
    )

archipy.configs.config_template.FastAPIConfig.PROJECT_NAME class-attribute instance-attribute 

PROJECT_NAME: str = Field(
    default="project_name",
    description="Name of the FastAPI project",
)

archipy.configs.config_template.FastAPIConfig.API_PREFIX class-attribute instance-attribute 

API_PREFIX: str = Field(
    default="/api",
    description="URL prefix for API endpoints",
)

archipy.configs.config_template.FastAPIConfig.ACCESS_LOG class-attribute instance-attribute 

ACCESS_LOG: bool = Field(
    default=True,
    description="Whether to enable access logging",
)

archipy.configs.config_template.FastAPIConfig.BACKLOG class-attribute instance-attribute 

BACKLOG: int = Field(
    default=2048,
    description="Maximum number of queued connections",
)

archipy.configs.config_template.FastAPIConfig.DATE_HEADER class-attribute instance-attribute 

DATE_HEADER: bool = Field(
    default=True,
    description="Whether to include date header in responses",
)

archipy.configs.config_template.FastAPIConfig.FORWARDED_ALLOW_IPS class-attribute instance-attribute 

FORWARDED_ALLOW_IPS: list[str] | None = Field(
    default=None,
    description="List of allowed forwarded IPs",
)

archipy.configs.config_template.FastAPIConfig.LIMIT_CONCURRENCY class-attribute instance-attribute 

LIMIT_CONCURRENCY: int | None = Field(
    default=None, description="Maximum concurrent requests"
)

archipy.configs.config_template.FastAPIConfig.LIMIT_MAX_REQUESTS class-attribute instance-attribute 

LIMIT_MAX_REQUESTS: int | None = Field(
    default=None, description="Maximum number of requests"
)

archipy.configs.config_template.FastAPIConfig.CORS_MIDDLEWARE_ALLOW_CREDENTIALS class-attribute instance-attribute 

CORS_MIDDLEWARE_ALLOW_CREDENTIALS: bool = Field(
    default=True,
    description="Whether to allow credentials in CORS",
)

archipy.configs.config_template.FastAPIConfig.CORS_MIDDLEWARE_ALLOW_HEADERS class-attribute instance-attribute 

CORS_MIDDLEWARE_ALLOW_HEADERS: list[str] = Field(
    default=["*"], description="Allowed CORS headers"
)

archipy.configs.config_template.FastAPIConfig.CORS_MIDDLEWARE_ALLOW_METHODS class-attribute instance-attribute 

CORS_MIDDLEWARE_ALLOW_METHODS: list[str] = Field(
    default=["*"], description="Allowed CORS methods"
)

archipy.configs.config_template.FastAPIConfig.CORS_MIDDLEWARE_ALLOW_ORIGINS class-attribute instance-attribute 

CORS_MIDDLEWARE_ALLOW_ORIGINS: list[str] = Field(
    default=["*"], description="Allowed CORS origins"
)

archipy.configs.config_template.FastAPIConfig.PROXY_HEADERS class-attribute instance-attribute 

PROXY_HEADERS: bool = Field(
    default=True,
    description="Whether to trust proxy headers",
)

archipy.configs.config_template.FastAPIConfig.RELOAD class-attribute instance-attribute 

RELOAD: bool = Field(
    default=False,
    description="Whether to enable auto-reload",
)

archipy.configs.config_template.FastAPIConfig.SERVER_HEADER class-attribute instance-attribute 

SERVER_HEADER: bool = Field(
    default=True,
    description="Whether to include server header",
)

archipy.configs.config_template.FastAPIConfig.SERVE_HOST class-attribute instance-attribute 

SERVE_HOST: str = Field(
    default="0.0.0.0",
    description="Host to serve the application on",
)

archipy.configs.config_template.FastAPIConfig.SERVE_PORT class-attribute instance-attribute 

SERVE_PORT: int = Field(
    default=8100,
    description="Port to serve the application on",
)

archipy.configs.config_template.FastAPIConfig.TIMEOUT_GRACEFUL_SHUTDOWN class-attribute instance-attribute 

TIMEOUT_GRACEFUL_SHUTDOWN: int | None = Field(
    default=None, description="Graceful shutdown timeout"
)

archipy.configs.config_template.FastAPIConfig.TIMEOUT_KEEP_ALIVE class-attribute instance-attribute 

TIMEOUT_KEEP_ALIVE: int = Field(
    default=5, description="Keep-alive timeout"
)

archipy.configs.config_template.FastAPIConfig.WORKERS_COUNT class-attribute instance-attribute 

WORKERS_COUNT: int = Field(
    default=4, description="Number of worker processes"
)

archipy.configs.config_template.FastAPIConfig.WS_MAX_SIZE class-attribute instance-attribute 

WS_MAX_SIZE: int = Field(
    default=16777216,
    description="Maximum WebSocket message size",
)

archipy.configs.config_template.FastAPIConfig.WS_PER_MESSAGE_DEFLATE class-attribute instance-attribute 

WS_PER_MESSAGE_DEFLATE: bool = Field(
    default=True,
    description="Whether to enable WebSocket compression",
)

archipy.configs.config_template.FastAPIConfig.WS_PING_INTERVAL class-attribute instance-attribute 

WS_PING_INTERVAL: float = Field(
    default=20.0, description="WebSocket ping interval"
)

archipy.configs.config_template.FastAPIConfig.WS_PING_TIMEOUT class-attribute instance-attribute 

WS_PING_TIMEOUT: float = Field(
    default=20.0, description="WebSocket ping timeout"
)

archipy.configs.config_template.FastAPIConfig.OPENAPI_URL class-attribute instance-attribute 

OPENAPI_URL: str | None = Field(
    default=None, description="URL for OpenAPI schema"
)

archipy.configs.config_template.FastAPIConfig.DOCS_URL class-attribute instance-attribute 

DOCS_URL: str | None = Field(
    default=None, description="URL for API documentation"
)

archipy.configs.config_template.FastAPIConfig.RE_DOC_URL class-attribute instance-attribute 

RE_DOC_URL: str | None = Field(
    default=None, description="URL for ReDoc documentation"
)

archipy.configs.config_template.FastAPIConfig.SWAGGER_UI_PARAMS class-attribute instance-attribute 

SWAGGER_UI_PARAMS: dict[str, str] | None = Field(
    default={"docExpansion": "none"},
    description="Swagger UI parameters",
)

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for gRPC services.

Controls gRPC server behavior, including connection parameters, performance tuning, and timeout settings.

Source code in archipy/configs/config_template.py 
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
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
class GrpcConfig(BaseModel):
    """Configuration settings for gRPC services.

    Controls gRPC server behavior, including connection parameters,
    performance tuning, and timeout settings.
    """

    SERVE_PORT: int = Field(default=8100, description="Port to serve gRPC on")
    SERVE_HOST: str = Field(default="[::]", description="Host to serve gRPC on")  # IPv6 equivalent of 0.0.0.0
    THREAD_WORKER_COUNT: int | None = Field(default=None, description="Number of worker threads")
    MAX_CONCURRENT_RPCS: int | None = Field(default=None, description="Maximum number of concurrent requests")
    THREAD_PER_CPU_CORE: int = Field(
        default=40,
        description="Threads per CPU core",
    )  # Adjust based on thread block to cpu time ratio
    SERVER_OPTIONS_CONFIG_LIST: list[tuple[str, int]] = Field(
        default=[
            ("grpc.max_metadata_size", 1 * 1024 * 1024),
            ("grpc.max_message_length", 128 * 1024 * 1024),
            ("grpc.max_receive_message_length", 128 * 1024 * 1024),
            ("grpc.max_send_message_length", 128 * 1024 * 1024),
            ("grpc.keepalive_time_ms", 5000),
            ("grpc.keepalive_timeout_ms", 1000),
            ("grpc.http2.min_ping_interval_without_data_ms", 5000),
            ("grpc.max_connection_idle_ms", 10000),
            ("grpc.max_connection_age_ms", 30000),
            ("grpc.max_connection_age_grace_ms", 5000),
            ("grpc.http2.max_pings_without_data", 0),
            ("grpc.keepalive_permit_without_calls", 1),
            ("grpc.http2.max_ping_strikes", 0),
            ("grpc.http2.min_recv_ping_interval_without_data_ms", 4000),
        ],
        description="Server configuration options",
    )

    STUB_OPTIONS_CONFIG_LIST: list[tuple[str, int | str]] = Field(
        default=[
            ("grpc.max_metadata_size", 1 * 1024 * 1024),
            ("grpc.max_message_length", 128 * 1024 * 1024),
            ("grpc.max_receive_message_length", 128 * 1024 * 1024),
            ("grpc.max_send_message_length", 128 * 1024 * 1024),
            ("grpc.keepalive_time_ms", 5000),
            ("grpc.keepalive_timeout_ms", 1000),
            ("grpc.http2.max_pings_without_data", 0),
            ("grpc.keepalive_permit_without_calls", 1),
            (
                "grpc.service_config",
                '{"methodConfig": [{"name": [],'
                ' "timeout": "1s", "waitForReady": true,'
                ' "retryPolicy": {"maxAttempts": 5,'
                ' "initialBackoff": "0.1s",'
                ' "maxBackoff": "1s",'
                ' "backoffMultiplier": 2,'
                ' "retryableStatusCodes": ["UNAVAILABLE", "ABORTED",'
                ' "RESOURCE_EXHAUSTED"]}}]}',
            ),
        ],
        description="Client stub configuration options",
    )

archipy.configs.config_template.GrpcConfig.SERVE_PORT class-attribute instance-attribute 

SERVE_PORT: int = Field(
    default=8100, description="Port to serve gRPC on"
)

archipy.configs.config_template.GrpcConfig.SERVE_HOST class-attribute instance-attribute 

SERVE_HOST: str = Field(
    default="[::]", description="Host to serve gRPC on"
)

archipy.configs.config_template.GrpcConfig.THREAD_WORKER_COUNT class-attribute instance-attribute 

THREAD_WORKER_COUNT: int | None = Field(
    default=None, description="Number of worker threads"
)

archipy.configs.config_template.GrpcConfig.MAX_CONCURRENT_RPCS class-attribute instance-attribute 

MAX_CONCURRENT_RPCS: int | None = Field(
    default=None,
    description="Maximum number of concurrent requests",
)

archipy.configs.config_template.GrpcConfig.THREAD_PER_CPU_CORE class-attribute instance-attribute 

THREAD_PER_CPU_CORE: int = Field(
    default=40, description="Threads per CPU core"
)

archipy.configs.config_template.GrpcConfig.SERVER_OPTIONS_CONFIG_LIST class-attribute instance-attribute 

SERVER_OPTIONS_CONFIG_LIST: list[tuple[str, int]] = Field(
    default=[
        ("grpc.max_metadata_size", 1 * 1024 * 1024),
        ("grpc.max_message_length", 128 * 1024 * 1024),
        (
            "grpc.max_receive_message_length",
            128 * 1024 * 1024,
        ),
        ("grpc.max_send_message_length", 128 * 1024 * 1024),
        ("grpc.keepalive_time_ms", 5000),
        ("grpc.keepalive_timeout_ms", 1000),
        (
            "grpc.http2.min_ping_interval_without_data_ms",
            5000,
        ),
        ("grpc.max_connection_idle_ms", 10000),
        ("grpc.max_connection_age_ms", 30000),
        ("grpc.max_connection_age_grace_ms", 5000),
        ("grpc.http2.max_pings_without_data", 0),
        ("grpc.keepalive_permit_without_calls", 1),
        ("grpc.http2.max_ping_strikes", 0),
        (
            "grpc.http2.min_recv_ping_interval_without_data_ms",
            4000,
        ),
    ],
    description="Server configuration options",
)

archipy.configs.config_template.GrpcConfig.STUB_OPTIONS_CONFIG_LIST class-attribute instance-attribute 

STUB_OPTIONS_CONFIG_LIST: list[tuple[str, int | str]] = (
    Field(
        default=[
            ("grpc.max_metadata_size", 1 * 1024 * 1024),
            ("grpc.max_message_length", 128 * 1024 * 1024),
            (
                "grpc.max_receive_message_length",
                128 * 1024 * 1024,
            ),
            (
                "grpc.max_send_message_length",
                128 * 1024 * 1024,
            ),
            ("grpc.keepalive_time_ms", 5000),
            ("grpc.keepalive_timeout_ms", 1000),
            ("grpc.http2.max_pings_without_data", 0),
            ("grpc.keepalive_permit_without_calls", 1),
            (
                "grpc.service_config",
                '{"methodConfig": [{"name": [], "timeout": "1s", "waitForReady": true, "retryPolicy": {"maxAttempts": 5, "initialBackoff": "0.1s", "maxBackoff": "1s", "backoffMultiplier": 2, "retryableStatusCodes": ["UNAVAILABLE", "ABORTED", "RESOURCE_EXHAUSTED"]}}]}',
            ),
        ],
        description="Client stub configuration options",
    )
)

options: show_root_toc_entry: false heading_level: 3

Observability Configs

Bases: BaseModel

Configuration settings for Sentry error tracking integration.

Controls Sentry client behavior, including DSN, sampling rates, and debug settings.

Source code in archipy/configs/config_template.py 
652
653
654
655
656
657
658
659
660
661
662
663
class SentryConfig(BaseModel):
    """Configuration settings for Sentry error tracking integration.

    Controls Sentry client behavior, including DSN, sampling rates, and debug settings.
    """

    IS_ENABLED: bool = Field(default=False, description="Whether Sentry is enabled")
    DSN: str | None = Field(default=None, description="Sentry DSN for error reporting")
    DEBUG: bool = Field(default=False, description="Whether to enable debug mode")
    RELEASE: str = Field(default="", description="Application release version")
    SAMPLE_RATE: float = Field(default=1.0, description="Error sampling rate (0.0 to 1.0)")
    TRACES_SAMPLE_RATE: float = Field(default=0.0, description="Performance monitoring sampling rate (0.0 to 1.0)")

archipy.configs.config_template.SentryConfig.IS_ENABLED class-attribute instance-attribute 

IS_ENABLED: bool = Field(
    default=False, description="Whether Sentry is enabled"
)

archipy.configs.config_template.SentryConfig.DSN class-attribute instance-attribute 

DSN: str | None = Field(
    default=None,
    description="Sentry DSN for error reporting",
)

archipy.configs.config_template.SentryConfig.DEBUG class-attribute instance-attribute 

DEBUG: bool = Field(
    default=False,
    description="Whether to enable debug mode",
)

archipy.configs.config_template.SentryConfig.RELEASE class-attribute instance-attribute 

RELEASE: str = Field(
    default="", description="Application release version"
)

archipy.configs.config_template.SentryConfig.SAMPLE_RATE class-attribute instance-attribute 

SAMPLE_RATE: float = Field(
    default=1.0,
    description="Error sampling rate (0.0 to 1.0)",
)

archipy.configs.config_template.SentryConfig.TRACES_SAMPLE_RATE class-attribute instance-attribute 

TRACES_SAMPLE_RATE: float = Field(
    default=0.0,
    description="Performance monitoring sampling rate (0.0 to 1.0)",
)

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for Prometheus metrics integration.

Controls whether Prometheus metrics collection is enabled and the port for the metrics endpoint.

Source code in archipy/configs/config_template.py 
585
586
587
588
589
590
591
592
593
class PrometheusConfig(BaseModel):
    """Configuration settings for Prometheus metrics integration.

    Controls whether Prometheus metrics collection is enabled and the port
    for the metrics endpoint.
    """

    IS_ENABLED: bool = Field(default=False, description="Whether Prometheus metrics are enabled")
    SERVER_PORT: int = Field(default=8200, description="Port for the Prometheus metrics endpoint")

archipy.configs.config_template.PrometheusConfig.IS_ENABLED class-attribute instance-attribute 

IS_ENABLED: bool = Field(
    default=False,
    description="Whether Prometheus metrics are enabled",
)

archipy.configs.config_template.PrometheusConfig.SERVER_PORT class-attribute instance-attribute 

SERVER_PORT: int = Field(
    default=8200,
    description="Port for the Prometheus metrics endpoint",
)

options: show_root_toc_entry: false heading_level: 3

Payment Configs

Bases: BaseModel

Configuration settings for Parsian Shaparak payment gateway integration.

Controls connection parameters and authentication for the Parsian Shaparak payment gateway services.

Source code in archipy/configs/config_template.py 
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
class ParsianShaparakConfig(BaseModel):
    """Configuration settings for Parsian Shaparak payment gateway integration.

    Controls connection parameters and authentication for the Parsian Shaparak
    payment gateway services.
    """

    LOGIN_ACCOUNT: str | None = Field(default=None, description="Merchant login account for authentication")
    PAYMENT_WSDL_URL: str = Field(
        default="https://pec.shaparak.ir/NewIPGServices/Sale/SaleService.asmx?WSDL",
        description="WSDL URL for the payment service",
    )
    CONFIRM_WSDL_URL: str = Field(
        default="https://pec.shaparak.ir/NewIPGServices/Confirm/ConfirmService.asmx?WSDL",
        description="WSDL URL for the confirm service",
    )
    REVERSAL_WSDL_URL: str = Field(
        default="https://pec.shaparak.ir/NewIPGServices/Reverse/ReversalService.asmx?WSDL",
        description="WSDL URL for the reversal service",
    )
    PROXIES: dict[str, str] | None = Field(
        default=None,
        description="Optional HTTP/HTTPS proxy configuration dictionary",
    )

archipy.configs.config_template.ParsianShaparakConfig.LOGIN_ACCOUNT class-attribute instance-attribute 

LOGIN_ACCOUNT: str | None = Field(
    default=None,
    description="Merchant login account for authentication",
)

archipy.configs.config_template.ParsianShaparakConfig.PAYMENT_WSDL_URL class-attribute instance-attribute 

PAYMENT_WSDL_URL: str = Field(
    default="https://pec.shaparak.ir/NewIPGServices/Sale/SaleService.asmx?WSDL",
    description="WSDL URL for the payment service",
)

archipy.configs.config_template.ParsianShaparakConfig.CONFIRM_WSDL_URL class-attribute instance-attribute 

CONFIRM_WSDL_URL: str = Field(
    default="https://pec.shaparak.ir/NewIPGServices/Confirm/ConfirmService.asmx?WSDL",
    description="WSDL URL for the confirm service",
)

archipy.configs.config_template.ParsianShaparakConfig.REVERSAL_WSDL_URL class-attribute instance-attribute 

REVERSAL_WSDL_URL: str = Field(
    default="https://pec.shaparak.ir/NewIPGServices/Reverse/ReversalService.asmx?WSDL",
    description="WSDL URL for the reversal service",
)

archipy.configs.config_template.ParsianShaparakConfig.PROXIES class-attribute instance-attribute 

PROXIES: dict[str, str] | None = Field(
    default=None,
    description="Optional HTTP/HTTPS proxy configuration dictionary",
)

options: show_root_toc_entry: false heading_level: 3

Application Configs

Bases: BaseModel

Configuration settings for authentication and security.

Controls JWT token settings, TOTP configuration, rate limiting, password policies, and token security features.

Source code in archipy/configs/config_template.py 
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
class AuthConfig(BaseModel):
    """Configuration settings for authentication and security.

    Controls JWT token settings, TOTP configuration, rate limiting,
    password policies, and token security features.
    """

    # JWT Settings
    SECRET_KEY: SecretStr | None = Field(default=None, description="JWT signing key")
    ACCESS_TOKEN_EXPIRES_IN: int = Field(
        default=1 * 60 * 60,
        description="Access token expiration in seconds",
    )  # 1 hour
    REFRESH_TOKEN_EXPIRES_IN: int = Field(
        default=24 * 60 * 60,
        description="Refresh token expiration in seconds",
    )  # 24 hours
    HASH_ALGORITHM: str = Field(default="HS256", description="JWT signing algorithm")
    JWT_ISSUER: str = Field(default="your-app-name", description="JWT issuer claim")
    JWT_AUDIENCE: str = Field(default="your-app-audience", description="JWT audience claim")
    TOKEN_VERSION: int = Field(default=1, description="JWT token version")

    # TOTP Settings
    TOTP_SECRET_KEY: SecretStr | None = Field(default=None, description="TOTP master key")
    TOTP_HASH_ALGORITHM: str = Field(
        default="SHA1",
        description="Hash algorithm for TOTP generation (SHA1, SHA256, SHA512)",
    )
    TOTP_LENGTH: int = Field(default=6, ge=6, le=8, description="TOTP code length")
    TOTP_EXPIRES_IN: int = Field(default=300, description="TOTP expiration time in seconds (5 minutes)")
    TOTP_TIME_STEP: int = Field(default=30, description="TOTP time step in seconds")
    TOTP_VERIFICATION_WINDOW: int = Field(default=1, description="Number of time steps to check before/after")
    TOTP_MAX_ATTEMPTS: int = Field(default=3, description="Maximum failed TOTP attempts before lockout")
    TOTP_LOCKOUT_TIME: int = Field(default=300, description="Lockout time in seconds after max attempts")

    # Rate Limiting Settings
    LOGIN_RATE_LIMIT: int = Field(default=5, description="Maximum login attempts per minute")
    TOTP_RATE_LIMIT: int = Field(default=3, description="Maximum TOTP requests per minute")
    PASSWORD_RESET_RATE_LIMIT: int = Field(default=3, description="Maximum password reset requests per hour")

    # Password Policy
    HASH_ITERATIONS: int = Field(default=100000, description="Password hash iterations")
    MIN_LENGTH: int = Field(default=12, ge=8, description="Minimum password length")
    REQUIRE_DIGIT: bool = Field(default=True, description="Whether password requires digits")
    REQUIRE_LOWERCASE: bool = Field(default=True, description="Whether password requires lowercase")
    REQUIRE_SPECIAL: bool = Field(default=True, description="Whether password requires special chars")
    REQUIRE_UPPERCASE: bool = Field(default=True, description="Whether password requires uppercase")
    SALT_LENGTH: int = Field(default=16, description="Password salt length")
    SPECIAL_CHARACTERS: set[str] = Field(default=set("!@#$%^&*()-_+="), description="Set of allowed special characters")
    PASSWORD_HISTORY_SIZE: int = Field(default=3, description="Number of previous passwords to remember")

    # Token Security
    ENABLE_JTI_CLAIM: bool = Field(default=True, description="Enable JWT ID claim for token tracking")
    ENABLE_TOKEN_ROTATION: bool = Field(default=True, description="Enable refresh token rotation")
    REFRESH_TOKEN_REUSE_INTERVAL: int = Field(default=60, description="Grace period for refresh token reuse in seconds")

archipy.configs.config_template.AuthConfig.SECRET_KEY class-attribute instance-attribute 

SECRET_KEY: SecretStr | None = Field(
    default=None, description="JWT signing key"
)

archipy.configs.config_template.AuthConfig.ACCESS_TOKEN_EXPIRES_IN class-attribute instance-attribute 

ACCESS_TOKEN_EXPIRES_IN: int = Field(
    default=1 * 60 * 60,
    description="Access token expiration in seconds",
)

archipy.configs.config_template.AuthConfig.REFRESH_TOKEN_EXPIRES_IN class-attribute instance-attribute 

REFRESH_TOKEN_EXPIRES_IN: int = Field(
    default=24 * 60 * 60,
    description="Refresh token expiration in seconds",
)

archipy.configs.config_template.AuthConfig.HASH_ALGORITHM class-attribute instance-attribute 

HASH_ALGORITHM: str = Field(
    default="HS256", description="JWT signing algorithm"
)

archipy.configs.config_template.AuthConfig.JWT_ISSUER class-attribute instance-attribute 

JWT_ISSUER: str = Field(
    default="your-app-name", description="JWT issuer claim"
)

archipy.configs.config_template.AuthConfig.JWT_AUDIENCE class-attribute instance-attribute 

JWT_AUDIENCE: str = Field(
    default="your-app-audience",
    description="JWT audience claim",
)

archipy.configs.config_template.AuthConfig.TOKEN_VERSION class-attribute instance-attribute 

TOKEN_VERSION: int = Field(
    default=1, description="JWT token version"
)

archipy.configs.config_template.AuthConfig.TOTP_SECRET_KEY class-attribute instance-attribute 

TOTP_SECRET_KEY: SecretStr | None = Field(
    default=None, description="TOTP master key"
)

archipy.configs.config_template.AuthConfig.TOTP_HASH_ALGORITHM class-attribute instance-attribute 

TOTP_HASH_ALGORITHM: str = Field(
    default="SHA1",
    description="Hash algorithm for TOTP generation (SHA1, SHA256, SHA512)",
)

archipy.configs.config_template.AuthConfig.TOTP_LENGTH class-attribute instance-attribute 

TOTP_LENGTH: int = Field(
    default=6, ge=6, le=8, description="TOTP code length"
)

archipy.configs.config_template.AuthConfig.TOTP_EXPIRES_IN class-attribute instance-attribute 

TOTP_EXPIRES_IN: int = Field(
    default=300,
    description="TOTP expiration time in seconds (5 minutes)",
)

archipy.configs.config_template.AuthConfig.TOTP_TIME_STEP class-attribute instance-attribute 

TOTP_TIME_STEP: int = Field(
    default=30, description="TOTP time step in seconds"
)

archipy.configs.config_template.AuthConfig.TOTP_VERIFICATION_WINDOW class-attribute instance-attribute 

TOTP_VERIFICATION_WINDOW: int = Field(
    default=1,
    description="Number of time steps to check before/after",
)

archipy.configs.config_template.AuthConfig.TOTP_MAX_ATTEMPTS class-attribute instance-attribute 

TOTP_MAX_ATTEMPTS: int = Field(
    default=3,
    description="Maximum failed TOTP attempts before lockout",
)

archipy.configs.config_template.AuthConfig.TOTP_LOCKOUT_TIME class-attribute instance-attribute 

TOTP_LOCKOUT_TIME: int = Field(
    default=300,
    description="Lockout time in seconds after max attempts",
)

archipy.configs.config_template.AuthConfig.LOGIN_RATE_LIMIT class-attribute instance-attribute 

LOGIN_RATE_LIMIT: int = Field(
    default=5,
    description="Maximum login attempts per minute",
)

archipy.configs.config_template.AuthConfig.TOTP_RATE_LIMIT class-attribute instance-attribute 

TOTP_RATE_LIMIT: int = Field(
    default=3,
    description="Maximum TOTP requests per minute",
)

archipy.configs.config_template.AuthConfig.PASSWORD_RESET_RATE_LIMIT class-attribute instance-attribute 

PASSWORD_RESET_RATE_LIMIT: int = Field(
    default=3,
    description="Maximum password reset requests per hour",
)

archipy.configs.config_template.AuthConfig.HASH_ITERATIONS class-attribute instance-attribute 

HASH_ITERATIONS: int = Field(
    default=100000, description="Password hash iterations"
)

archipy.configs.config_template.AuthConfig.MIN_LENGTH class-attribute instance-attribute 

MIN_LENGTH: int = Field(
    default=12, ge=8, description="Minimum password length"
)

archipy.configs.config_template.AuthConfig.REQUIRE_DIGIT class-attribute instance-attribute 

REQUIRE_DIGIT: bool = Field(
    default=True,
    description="Whether password requires digits",
)

archipy.configs.config_template.AuthConfig.REQUIRE_LOWERCASE class-attribute instance-attribute 

REQUIRE_LOWERCASE: bool = Field(
    default=True,
    description="Whether password requires lowercase",
)

archipy.configs.config_template.AuthConfig.REQUIRE_SPECIAL class-attribute instance-attribute 

REQUIRE_SPECIAL: bool = Field(
    default=True,
    description="Whether password requires special chars",
)

archipy.configs.config_template.AuthConfig.REQUIRE_UPPERCASE class-attribute instance-attribute 

REQUIRE_UPPERCASE: bool = Field(
    default=True,
    description="Whether password requires uppercase",
)

archipy.configs.config_template.AuthConfig.SALT_LENGTH class-attribute instance-attribute 

SALT_LENGTH: int = Field(
    default=16, description="Password salt length"
)

archipy.configs.config_template.AuthConfig.SPECIAL_CHARACTERS class-attribute instance-attribute 

SPECIAL_CHARACTERS: set[str] = Field(
    default=set("!@#$%^&*()-_+="),
    description="Set of allowed special characters",
)

archipy.configs.config_template.AuthConfig.PASSWORD_HISTORY_SIZE class-attribute instance-attribute 

PASSWORD_HISTORY_SIZE: int = Field(
    default=3,
    description="Number of previous passwords to remember",
)

archipy.configs.config_template.AuthConfig.ENABLE_JTI_CLAIM class-attribute instance-attribute 

ENABLE_JTI_CLAIM: bool = Field(
    default=True,
    description="Enable JWT ID claim for token tracking",
)

archipy.configs.config_template.AuthConfig.ENABLE_TOKEN_ROTATION class-attribute instance-attribute 

ENABLE_TOKEN_ROTATION: bool = Field(
    default=True,
    description="Enable refresh token rotation",
)

archipy.configs.config_template.AuthConfig.REFRESH_TOKEN_REUSE_INTERVAL class-attribute instance-attribute 

REFRESH_TOKEN_REUSE_INTERVAL: int = Field(
    default=60,
    description="Grace period for refresh token reuse in seconds",
)

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for file handling capabilities.

Controls file link security, expiration policies, and file type restrictions.

Source code in archipy/configs/config_template.py 
740
741
742
743
744
745
746
747
748
749
750
751
752
class FileConfig(BaseModel):
    """Configuration settings for file handling capabilities.

    Controls file link security, expiration policies, and file type restrictions.
    """

    SECRET_KEY: str | None = Field(default=None, description="Secret key used for generating secure file links")
    DEFAULT_EXPIRY_MINUTES: int = Field(
        default=60,
        ge=1,
        description="Default number of minutes until link expiration",  # Default 60 minutes (1 hour)
    )
    ALLOWED_EXTENSIONS: list[str] = Field(default=["jpg", "jpeg", "png"], description="List of allowed file extensions")

archipy.configs.config_template.FileConfig.SECRET_KEY class-attribute instance-attribute 

SECRET_KEY: str | None = Field(
    default=None,
    description="Secret key used for generating secure file links",
)

archipy.configs.config_template.FileConfig.DEFAULT_EXPIRY_MINUTES class-attribute instance-attribute 

DEFAULT_EXPIRY_MINUTES: int = Field(
    default=60,
    ge=1,
    description="Default number of minutes until link expiration",
)

archipy.configs.config_template.FileConfig.ALLOWED_EXTENSIONS class-attribute instance-attribute 

ALLOWED_EXTENSIONS: list[str] = Field(
    default=["jpg", "jpeg", "png"],
    description="List of allowed file extensions",
)

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for date and time handling.

Controls API connections for specialized date/time services and date caching behavior.

Source code in archipy/configs/config_template.py 
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
class DatetimeConfig(BaseModel):
    """Configuration settings for date and time handling.

    Controls API connections for specialized date/time services
    and date caching behavior.
    """

    TIME_IR_API_KEY: str | None = Field(
        default="ZAVdqwuySASubByCed5KYuYMzb9uB2f7",
        description="API key for time.ir service",
    )
    TIME_IR_API_ENDPOINT: str | None = Field(
        default="https://api.time.ir/v1/event/fa/events/calendar",
        description="Endpoint for time.ir service",
    )
    REQUEST_TIMEOUT: int = Field(default=5, description="Request timeout in seconds")
    MAX_RETRIES: int = Field(default=3, description="Maximum retry attempts")
    CACHE_TTL: int = Field(default=86400, description="Cache time-to-live in seconds (24 hours)")
    HISTORICAL_CACHE_TTL: int = Field(
        default=604800,
        description="Cache time-to-live for historical dates in seconds (7 days)",
    )

archipy.configs.config_template.DatetimeConfig.TIME_IR_API_KEY class-attribute instance-attribute 

TIME_IR_API_KEY: str | None = Field(
    default="ZAVdqwuySASubByCed5KYuYMzb9uB2f7",
    description="API key for time.ir service",
)

archipy.configs.config_template.DatetimeConfig.TIME_IR_API_ENDPOINT class-attribute instance-attribute 

TIME_IR_API_ENDPOINT: str | None = Field(
    default="https://api.time.ir/v1/event/fa/events/calendar",
    description="Endpoint for time.ir service",
)

archipy.configs.config_template.DatetimeConfig.REQUEST_TIMEOUT class-attribute instance-attribute 

REQUEST_TIMEOUT: int = Field(
    default=5, description="Request timeout in seconds"
)

archipy.configs.config_template.DatetimeConfig.MAX_RETRIES class-attribute instance-attribute 

MAX_RETRIES: int = Field(
    default=3, description="Maximum retry attempts"
)

archipy.configs.config_template.DatetimeConfig.CACHE_TTL class-attribute instance-attribute 

CACHE_TTL: int = Field(
    default=86400,
    description="Cache time-to-live in seconds (24 hours)",
)

archipy.configs.config_template.DatetimeConfig.HISTORICAL_CACHE_TTL class-attribute instance-attribute 

HISTORICAL_CACHE_TTL: int = Field(
    default=604800,
    description="Cache time-to-live for historical dates in seconds (7 days)",
)

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for Temporal workflow engine integration.

Controls connection parameters, security settings, and timeout configurations for Temporal workflow orchestration services.

Source code in archipy/configs/config_template.py 
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
class TemporalConfig(BaseModel):
    """Configuration settings for Temporal workflow engine integration.

    Controls connection parameters, security settings, and timeout configurations
    for Temporal workflow orchestration services.
    """

    HOST: str = Field(default="localhost", description="Temporal server host address")
    PORT: int = Field(default=7233, ge=1, le=65535, description="Temporal server port number")
    NAMESPACE: str = Field(default="default", description="Temporal namespace for workflow isolation")
    TASK_QUEUE: str = Field(default="task-queue", description="Default task queue name")

    # Metrics Configuration
    ENABLE_METRICS: bool = Field(
        default=False,
        description="Enable Prometheus metrics collection for Temporal workflows and activities",
    )
    METRICS_PORT: int = Field(
        default=8201,
        ge=1,
        le=65535,
        description="Port for Temporal Prometheus metrics endpoint (separate from main Prometheus port)",
    )

    # TLS Configuration
    TLS_CA_CERT: str | None = Field(default=None, description="Path to TLS CA certificate")
    TLS_CLIENT_CERT: str | None = Field(default=None, description="Path to TLS client certificate")
    TLS_CLIENT_KEY: str | None = Field(default=None, description="Path to TLS client private key")

    # Workflow Timeout Settings
    WORKFLOW_EXECUTION_TIMEOUT: int = Field(
        default=300,
        ge=1,
        description="Maximum workflow execution time in seconds",
    )
    WORKFLOW_RUN_TIMEOUT: int = Field(
        default=60,
        ge=1,
        description="Maximum single workflow run time in seconds",
    )
    WORKFLOW_TASK_TIMEOUT: int = Field(
        default=30,
        ge=1,
        description="Maximum workflow task processing time in seconds",
    )

    # Activity Timeout Settings
    ACTIVITY_START_TO_CLOSE_TIMEOUT: int = Field(
        default=30,
        ge=1,
        description="Maximum activity execution time in seconds",
    )
    ACTIVITY_HEARTBEAT_TIMEOUT: int = Field(
        default=10,
        ge=1,
        description="Activity heartbeat timeout in seconds",
    )

    # Retry Configuration
    RETRY_MAXIMUM_ATTEMPTS: int = Field(
        default=3,
        ge=1,
        description="Maximum retry attempts for failed activities",
    )
    RETRY_BACKOFF_COEFFICIENT: float = Field(
        default=2.0,
        ge=1.0,
        description="Backoff multiplier for retry delays",
    )
    RETRY_MAXIMUM_INTERVAL: int = Field(
        default=60,
        ge=1,
        description="Maximum retry interval in seconds",
    )

    @model_validator(mode="after")
    def validate_tls_configuration(self) -> Self:
        """Validate TLS configuration consistency."""
        tls_fields = [self.TLS_CA_CERT, self.TLS_CLIENT_CERT, self.TLS_CLIENT_KEY]
        tls_provided = [field for field in tls_fields if field is not None]

        if len(tls_provided) > 0 and len(tls_provided) != 3:
            raise InvalidArgumentError()

        return self

    @model_validator(mode="after")
    def validate_timeout_hierarchy(self) -> Self:
        """Validate timeout configuration hierarchy."""
        if self.WORKFLOW_RUN_TIMEOUT >= self.WORKFLOW_EXECUTION_TIMEOUT:
            raise InvalidArgumentError()

        if self.WORKFLOW_TASK_TIMEOUT >= self.WORKFLOW_RUN_TIMEOUT:
            raise InvalidArgumentError()

        return self

archipy.configs.config_template.TemporalConfig.HOST class-attribute instance-attribute 

HOST: str = Field(
    default="localhost",
    description="Temporal server host address",
)

archipy.configs.config_template.TemporalConfig.PORT class-attribute instance-attribute 

PORT: int = Field(
    default=7233,
    ge=1,
    le=65535,
    description="Temporal server port number",
)

archipy.configs.config_template.TemporalConfig.NAMESPACE class-attribute instance-attribute 

NAMESPACE: str = Field(
    default="default",
    description="Temporal namespace for workflow isolation",
)

archipy.configs.config_template.TemporalConfig.TASK_QUEUE class-attribute instance-attribute 

TASK_QUEUE: str = Field(
    default="task-queue",
    description="Default task queue name",
)

archipy.configs.config_template.TemporalConfig.ENABLE_METRICS class-attribute instance-attribute 

ENABLE_METRICS: bool = Field(
    default=False,
    description="Enable Prometheus metrics collection for Temporal workflows and activities",
)

archipy.configs.config_template.TemporalConfig.METRICS_PORT class-attribute instance-attribute 

METRICS_PORT: int = Field(
    default=8201,
    ge=1,
    le=65535,
    description="Port for Temporal Prometheus metrics endpoint (separate from main Prometheus port)",
)

archipy.configs.config_template.TemporalConfig.TLS_CA_CERT class-attribute instance-attribute 

TLS_CA_CERT: str | None = Field(
    default=None, description="Path to TLS CA certificate"
)

archipy.configs.config_template.TemporalConfig.TLS_CLIENT_CERT class-attribute instance-attribute 

TLS_CLIENT_CERT: str | None = Field(
    default=None,
    description="Path to TLS client certificate",
)

archipy.configs.config_template.TemporalConfig.TLS_CLIENT_KEY class-attribute instance-attribute 

TLS_CLIENT_KEY: str | None = Field(
    default=None,
    description="Path to TLS client private key",
)

archipy.configs.config_template.TemporalConfig.WORKFLOW_EXECUTION_TIMEOUT class-attribute instance-attribute 

WORKFLOW_EXECUTION_TIMEOUT: int = Field(
    default=300,
    ge=1,
    description="Maximum workflow execution time in seconds",
)

archipy.configs.config_template.TemporalConfig.WORKFLOW_RUN_TIMEOUT class-attribute instance-attribute 

WORKFLOW_RUN_TIMEOUT: int = Field(
    default=60,
    ge=1,
    description="Maximum single workflow run time in seconds",
)

archipy.configs.config_template.TemporalConfig.WORKFLOW_TASK_TIMEOUT class-attribute instance-attribute 

WORKFLOW_TASK_TIMEOUT: int = Field(
    default=30,
    ge=1,
    description="Maximum workflow task processing time in seconds",
)

archipy.configs.config_template.TemporalConfig.ACTIVITY_START_TO_CLOSE_TIMEOUT class-attribute instance-attribute 

ACTIVITY_START_TO_CLOSE_TIMEOUT: int = Field(
    default=30,
    ge=1,
    description="Maximum activity execution time in seconds",
)

archipy.configs.config_template.TemporalConfig.ACTIVITY_HEARTBEAT_TIMEOUT class-attribute instance-attribute 

ACTIVITY_HEARTBEAT_TIMEOUT: int = Field(
    default=10,
    ge=1,
    description="Activity heartbeat timeout in seconds",
)

archipy.configs.config_template.TemporalConfig.RETRY_MAXIMUM_ATTEMPTS class-attribute instance-attribute 

RETRY_MAXIMUM_ATTEMPTS: int = Field(
    default=3,
    ge=1,
    description="Maximum retry attempts for failed activities",
)

archipy.configs.config_template.TemporalConfig.RETRY_BACKOFF_COEFFICIENT class-attribute instance-attribute 

RETRY_BACKOFF_COEFFICIENT: float = Field(
    default=2.0,
    ge=1.0,
    description="Backoff multiplier for retry delays",
)

archipy.configs.config_template.TemporalConfig.RETRY_MAXIMUM_INTERVAL class-attribute instance-attribute 

RETRY_MAXIMUM_INTERVAL: int = Field(
    default=60,
    ge=1,
    description="Maximum retry interval in seconds",
)

archipy.configs.config_template.TemporalConfig.validate_tls_configuration 

validate_tls_configuration() -> Self

Validate TLS configuration consistency.

Source code in archipy/configs/config_template.py 
880
881
882
883
884
885
886
887
888
889
@model_validator(mode="after")
def validate_tls_configuration(self) -> Self:
    """Validate TLS configuration consistency."""
    tls_fields = [self.TLS_CA_CERT, self.TLS_CLIENT_CERT, self.TLS_CLIENT_KEY]
    tls_provided = [field for field in tls_fields if field is not None]

    if len(tls_provided) > 0 and len(tls_provided) != 3:
        raise InvalidArgumentError()

    return self

archipy.configs.config_template.TemporalConfig.validate_timeout_hierarchy 

validate_timeout_hierarchy() -> Self

Validate timeout configuration hierarchy.

Source code in archipy/configs/config_template.py 
891
892
893
894
895
896
897
898
899
900
@model_validator(mode="after")
def validate_timeout_hierarchy(self) -> Self:
    """Validate timeout configuration hierarchy."""
    if self.WORKFLOW_RUN_TIMEOUT >= self.WORKFLOW_EXECUTION_TIMEOUT:
        raise InvalidArgumentError()

    if self.WORKFLOW_TASK_TIMEOUT >= self.WORKFLOW_RUN_TIMEOUT:
        raise InvalidArgumentError()

    return self

options: show_root_toc_entry: false heading_level: 3

Environment Type

archipy.configs.environment_type.EnvironmentType

Bases: StrEnum

Enum representing different application environments.

This enum defines the available environment types for an application and provides helper properties to check the current environment.

Attributes:

Name Type Description
PRODUCTION

Production environment
BETA

Beta testing environment (human testing)
ALPHA

Alpha testing environment (human testing)
TEST

Automated testing environment
DEV

Development environment
LOCAL

Local development environment

Examples:

>>> from archipy.configs.environment_type import EnvironmentType
>>>
>>> # Setting environment in configuration
>>> env = EnvironmentType.DEV
>>>
>>> # Checking environment type
>>> if env.is_production:
...     print("Running in production mode")
>>> elif env.is_dev:
...     print("Running in development mode")
>>>
>>> # Getting appropriate log level
>>> log_level = env.log_level
>>> print(f"Log level: {log_level}")
Source code in archipy/configs/environment_type.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
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
class EnvironmentType(StrEnum):
    """Enum representing different application environments.

    This enum defines the available environment types for an application
    and provides helper properties to check the current environment.

    Attributes:
        PRODUCTION: Production environment
        BETA: Beta testing environment (human testing)
        ALPHA: Alpha testing environment (human testing)
        TEST: Automated testing environment
        DEV: Development environment
        LOCAL: Local development environment

    Examples:
        >>> from archipy.configs.environment_type import EnvironmentType
        >>>
        >>> # Setting environment in configuration
        >>> env = EnvironmentType.DEV
        >>>
        >>> # Checking environment type
        >>> if env.is_production:
        ...     print("Running in production mode")
        >>> elif env.is_dev:
        ...     print("Running in development mode")
        >>>
        >>> # Getting appropriate log level
        >>> log_level = env.log_level
        >>> print(f"Log level: {log_level}")
    """

    PRODUCTION = "PRODUCTION"
    BETA = "BETA"  # human test
    ALPHA = "ALPHA"  # human test
    TEST = "TEST"  # automatic test
    DEV = "DEV"
    LOCAL = "LOCAL"

    @enum.property
    def is_local(self) -> bool:
        """Check if the environment is LOCAL.

        Returns:
            bool: True if environment is LOCAL, False otherwise.
        """
        return self == self.LOCAL

    @enum.property
    def is_dev(self) -> bool:
        """Check if the environment is DEV.

        Returns:
            bool: True if environment is DEV, False otherwise.
        """
        return self == self.DEV

    @enum.property
    def is_test(self) -> bool:
        """Check if the environment is a testing environment.

        Returns:
            bool: True if environment is BETA, ALPHA, or TEST, False otherwise.
        """
        return self in (self.BETA, self.ALPHA, self.TEST)

    @enum.property
    def is_production(self) -> bool:
        """Check if the environment is a production environment.

        This returns True for PRODUCTION and False for all other environments.

        Returns:
            bool: True if not a test, dev, or local environment, False otherwise.
        """
        return not self.is_test and not self.is_dev and not self.is_local

    @enum.property
    def log_level(self) -> int:
        """Get the appropriate logging level for this environment.

        Returns:
            int: WARNING for production, INFO for test environments,
                DEBUG for development and local environments.
        """
        if self.is_production:
            return WARNING
        if self.is_test:
            return INFO
        return DEBUG

archipy.configs.environment_type.EnvironmentType.PRODUCTION class-attribute instance-attribute 

PRODUCTION = 'PRODUCTION'

archipy.configs.environment_type.EnvironmentType.BETA class-attribute instance-attribute 

BETA = 'BETA'

archipy.configs.environment_type.EnvironmentType.ALPHA class-attribute instance-attribute 

ALPHA = 'ALPHA'

archipy.configs.environment_type.EnvironmentType.TEST class-attribute instance-attribute 

TEST = 'TEST'

archipy.configs.environment_type.EnvironmentType.DEV class-attribute instance-attribute 

DEV = 'DEV'

archipy.configs.environment_type.EnvironmentType.LOCAL class-attribute instance-attribute 

LOCAL = 'LOCAL'

archipy.configs.environment_type.EnvironmentType.is_local 

is_local() -> bool

Check if the environment is LOCAL.

Returns:

Name Type Description
bool bool

True if environment is LOCAL, False otherwise.
Source code in archipy/configs/environment_type.py 
44
45
46
47
48
49
50
51
@enum.property
def is_local(self) -> bool:
    """Check if the environment is LOCAL.

    Returns:
        bool: True if environment is LOCAL, False otherwise.
    """
    return self == self.LOCAL

archipy.configs.environment_type.EnvironmentType.is_dev 

is_dev() -> bool

Check if the environment is DEV.

Returns:

Name Type Description
bool bool

True if environment is DEV, False otherwise.
Source code in archipy/configs/environment_type.py 
53
54
55
56
57
58
59
60
@enum.property
def is_dev(self) -> bool:
    """Check if the environment is DEV.

    Returns:
        bool: True if environment is DEV, False otherwise.
    """
    return self == self.DEV

archipy.configs.environment_type.EnvironmentType.is_test 

is_test() -> bool

Check if the environment is a testing environment.

Returns:

Name Type Description
bool bool

True if environment is BETA, ALPHA, or TEST, False otherwise.
Source code in archipy/configs/environment_type.py 
62
63
64
65
66
67
68
69
@enum.property
def is_test(self) -> bool:
    """Check if the environment is a testing environment.

    Returns:
        bool: True if environment is BETA, ALPHA, or TEST, False otherwise.
    """
    return self in (self.BETA, self.ALPHA, self.TEST)

archipy.configs.environment_type.EnvironmentType.is_production 

is_production() -> bool

Check if the environment is a production environment.

This returns True for PRODUCTION and False for all other environments.

Returns:

Name Type Description
bool bool

True if not a test, dev, or local environment, False otherwise.
Source code in archipy/configs/environment_type.py 
71
72
73
74
75
76
77
78
79
80
@enum.property
def is_production(self) -> bool:
    """Check if the environment is a production environment.

    This returns True for PRODUCTION and False for all other environments.

    Returns:
        bool: True if not a test, dev, or local environment, False otherwise.
    """
    return not self.is_test and not self.is_dev and not self.is_local

archipy.configs.environment_type.EnvironmentType.log_level 

log_level() -> int

Get the appropriate logging level for this environment.

Returns:

Name Type Description
int int

WARNING for production, INFO for test environments, DEBUG for development and local environments.
Source code in archipy/configs/environment_type.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
@enum.property
def log_level(self) -> int:
    """Get the appropriate logging level for this environment.

    Returns:
        int: WARNING for production, INFO for test environments,
            DEBUG for development and local environments.
    """
    if self.is_production:
        return WARNING
    if self.is_test:
        return INFO
    return DEBUG

options: show_root_toc_entry: false heading_level: 3 show_bases: true

Source Code

📁 Location: archipy/configs/

🔗 Browse Source