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 is built-in; override the default None for this service
    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

Base configuration system for ArchiPy applications.

Settings are loaded from multiple sources, applied in the following priority order (highest priority first):

  1. Docker/K8s secret files (file_secret_settings)
  2. HashiCorp Vault KV v2 secrets (when VAULT__ENABLED=true)
  3. pyproject.toml [tool.configs] section
  4. configs.toml or other specified TOML file
  5. OS-level environment variables
  6. .env file
  7. Default class field values / init settings

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. Docker/K8s secret files (file_secret_settings)
  2. HashiCorp Vault KV v2 secrets (when VAULT__ENABLED=true)
  3. pyproject.toml [tool.configs] section
  4. configs.toml or other specified TOML files
  5. OS-level environment variables
  6. Environment variables (.env file)
  7. Default class field values / init settings

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
APP_NAME str | None

Application identity; synced into nested configs in customize()
AUTH AuthConfig

Authentication and security settings
DATETIME DatetimeConfig

Date/time handling configuration
ELASTIC ElasticsearchConfig

Elasticsearch 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
GRPC_RATE_LIMIT GrpcRateLimitConfig

gRPC server rate limiting settings
KAFKA KafkaConfig

Kafka integration configuration
KEYCLOAK KeycloakConfig

Keycloak integration configuration
LANGUAGE LanguageType

Application default language
MINIO MinioConfig

MinIO object storage configuration
MYSQL_SQLALCHEMY MySQLSQLAlchemyConfig

MySQL SQLAlchemy configuration
OTEL OpentelemetryConfig

OpenTelemetry observability configuration
PARSIAN_SHAPARAK ParsianShaparakConfig

Parsian Shaparak payment gateway configuration
POSTGRES_SQLALCHEMY PostgresSQLAlchemyConfig

PostgreSQL SQLAlchemy configuration
REDIS RedisConfig

Redis cache configuration
SAMAN_SHAPARAK SamanShaparakConfig

Saman Shaparak (SEP) payment gateway configuration
SCYLLADB ScyllaDBConfig

ScyllaDB/Cassandra database configuration
SQLALCHEMY SQLAlchemyConfig

Database ORM configuration
SQLITE_SQLALCHEMY SqliteSQLAlchemyConfig

SQLite SQLAlchemy configuration
STARROCKS_SQLALCHEMY StarrocksSQLAlchemyConfig

Starrocks SQLAlchemy configuration
TEMPORAL TemporalConfig

Temporal workflow orchestration configuration
VAULT VaultConfig

HashiCorp Vault connection and secrets settings

Examples:

>>> from archipy.configs.base_config import BaseConfig
>>>
>>> class MyAppConfig(BaseConfig):
...     # Override defaults (APP_NAME is a built-in field)
...     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 
 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
223
224
225
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
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. Docker/K8s secret files (`file_secret_settings`)
    2. HashiCorp Vault KV v2 secrets (when `VAULT__ENABLED=true`)
    3. pyproject.toml [tool.configs] section
    4. configs.toml or other specified TOML files
    5. OS-level environment variables
    6. Environment variables (.env file)
    7. Default class field values / init settings

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

    Attributes:
        APP_NAME (str | None): Application identity; synced into nested configs in ``customize()``
        AUTH (AuthConfig): Authentication and security settings
        DATETIME (DatetimeConfig): Date/time handling configuration
        ELASTIC (ElasticsearchConfig): Elasticsearch 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
        GRPC_RATE_LIMIT (GrpcRateLimitConfig): gRPC server rate limiting settings
        KAFKA (KafkaConfig): Kafka integration configuration
        KEYCLOAK (KeycloakConfig): Keycloak integration configuration
        LANGUAGE (LanguageType): Application default language
        MINIO (MinioConfig): MinIO object storage configuration
        MYSQL_SQLALCHEMY (MySQLSQLAlchemyConfig): MySQL SQLAlchemy configuration
        OTEL (OpentelemetryConfig): OpenTelemetry observability configuration
        PARSIAN_SHAPARAK (ParsianShaparakConfig): Parsian Shaparak payment gateway configuration
        POSTGRES_SQLALCHEMY (PostgresSQLAlchemyConfig): PostgreSQL SQLAlchemy configuration
        REDIS (RedisConfig): Redis cache configuration
        SAMAN_SHAPARAK (SamanShaparakConfig): Saman Shaparak (SEP) payment gateway configuration
        SCYLLADB (ScyllaDBConfig): ScyllaDB/Cassandra database configuration
        SQLALCHEMY (SQLAlchemyConfig): Database ORM configuration
        SQLITE_SQLALCHEMY (SqliteSQLAlchemyConfig): SQLite SQLAlchemy configuration
        STARROCKS_SQLALCHEMY (StarrocksSQLAlchemyConfig): Starrocks SQLAlchemy configuration
        TEMPORAL (TemporalConfig): Temporal workflow orchestration configuration
        VAULT (VaultConfig): HashiCorp Vault connection and secrets settings

    Examples:
        >>> from archipy.configs.base_config import BaseConfig
        >>>
        >>> class MyAppConfig(BaseConfig):
        ...     # Override defaults (APP_NAME is a built-in field)
        ...     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

        Raises:
            ConfigurationError: If Vault is enabled but the ``vault`` extra is not installed.
        """
        sources: list[PydanticBaseSettingsSource] = [file_secret_settings]

        # Import Vault settings source only when explicitly enabled — avoids loading
        # hvac on cold start for apps that ship with the vault extra but leave it off.
        vault_enabled = os.environ.get("VAULT__ENABLED", "").strip().lower() in {
            "1",
            "true",
            "yes",
            "on",
        }
        if vault_enabled:
            try:
                vault_module = importlib.import_module("archipy.configs.vault_settings_source")
            except ImportError as e:
                raise ConfigurationError(
                    operation="vault_import",
                    reason='hvac is not installed; install with uv add "archipy[vault]"',
                ) from e
            sources.append(vault_module.VaultSettingsSource(settings_cls))

        sources.extend(
            [
                PyprojectTomlConfigSettingsSource(settings_cls),
                TomlConfigSettingsSource(settings_cls),
                env_settings,
                dotenv_settings,
                init_settings,
            ],
        )
        return tuple(sources)

    APP_NAME: str | None = None
    AUTH: AuthConfig = AuthConfig()
    DATETIME: DatetimeConfig = DatetimeConfig()
    ELASTIC: ElasticsearchConfig = ElasticsearchConfig()
    EMAIL: EmailConfig = EmailConfig()
    ENVIRONMENT: EnvironmentType = EnvironmentType.LOCAL
    FASTAPI: FastAPIConfig = FastAPIConfig()
    FILE: FileConfig = FileConfig()
    GRPC: GrpcConfig = GrpcConfig()
    GRPC_RATE_LIMIT: GrpcRateLimitConfig = GrpcRateLimitConfig()
    KAFKA: KafkaConfig = KafkaConfig()
    KEYCLOAK: KeycloakConfig = KeycloakConfig()
    MINIO: MinioConfig = MinioConfig()
    OTEL: OpentelemetryConfig = OpentelemetryConfig()
    PARSIAN_SHAPARAK: ParsianShaparakConfig = ParsianShaparakConfig()
    SAMAN_SHAPARAK: SamanShaparakConfig = SamanShaparakConfig()
    REDIS: RedisConfig = RedisConfig()
    SCYLLADB: ScyllaDBConfig = ScyllaDBConfig()
    SQLALCHEMY: SQLAlchemyConfig = SQLAlchemyConfig()
    STARROCKS_SQLALCHEMY: StarRocksSQLAlchemyConfig = StarRocksSQLAlchemyConfig()
    POSTGRES_SQLALCHEMY: PostgresSQLAlchemyConfig = PostgresSQLAlchemyConfig()
    MYSQL_SQLALCHEMY: MySQLSQLAlchemyConfig = MySQLSQLAlchemyConfig()
    SQLITE_SQLALCHEMY: SQLiteSQLAlchemyConfig = SQLiteSQLAlchemyConfig()
    TEMPORAL: TemporalConfig = TemporalConfig()
    VAULT: VaultConfig = VaultConfig()
    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.
        It is called automatically by `set_global()`.

        When ``APP_NAME`` is set, unset or placeholder nested identity fields are
        filled: ``OTEL.SERVICE_NAME``, ``FASTAPI.PROJECT_NAME``, ``AUTH.JWT_ISSUER``,
        and ``TEMPORAL.CLIENT_IDENTITY``. Explicit nested values always win.

        Examples:
            >>> class MyAppConfig(BaseConfig):
            ...     def customize(self) -> None:
            ...         super().customize()
            ...         self.REDIS.MASTER_HOST = "redis.internal"
            >>> config = MyAppConfig()
            >>> BaseConfig.set_global(config)  # calls customize() automatically
        """
        if self.OTEL.ENVIRONMENT is None:
            self.OTEL.ENVIRONMENT = self.ENVIRONMENT

        if self.APP_NAME:
            if self.OTEL.SERVICE_NAME is None:
                self.OTEL.SERVICE_NAME = self.APP_NAME
            if self.FASTAPI.PROJECT_NAME == "project_name":
                self.FASTAPI.PROJECT_NAME = self.APP_NAME
            if self.AUTH.JWT_ISSUER == "your-app-name":
                self.AUTH.JWT_ISSUER = self.APP_NAME
            if self.TEMPORAL.CLIENT_IDENTITY is None:
                self.TEMPORAL.CLIENT_IDENTITY = self.APP_NAME

    @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
        # Pydantic wraps unset private attrs in ModelPrivateAttr; ``is None`` alone never fires.
        if global_config is None or isinstance(global_config, ModelPrivateAttr):
            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.APP_NAME class-attribute instance-attribute 

APP_NAME: str | None = None

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.EMAIL class-attribute instance-attribute 

EMAIL: EmailConfig = EmailConfig()

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

ENVIRONMENT: EnvironmentType = 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.GRPC_RATE_LIMIT class-attribute instance-attribute 

GRPC_RATE_LIMIT: GrpcRateLimitConfig = GrpcRateLimitConfig()

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.OTEL class-attribute instance-attribute 

OTEL: OpentelemetryConfig = OpentelemetryConfig()

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

PARSIAN_SHAPARAK: ParsianShaparakConfig = (
    ParsianShaparakConfig()
)

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

SAMAN_SHAPARAK: SamanShaparakConfig = SamanShaparakConfig()

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.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.MYSQL_SQLALCHEMY class-attribute instance-attribute 

MYSQL_SQLALCHEMY: MySQLSQLAlchemyConfig = (
    MySQLSQLAlchemyConfig()
)

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.VAULT class-attribute instance-attribute 

VAULT: VaultConfig = VaultConfig()

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

LANGUAGE: LanguageType = 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

Raises:

Type Description
ConfigurationError

If Vault is enabled but the vault extra is not installed.
Source code in archipy/configs/base_config.py 
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
@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

    Raises:
        ConfigurationError: If Vault is enabled but the ``vault`` extra is not installed.
    """
    sources: list[PydanticBaseSettingsSource] = [file_secret_settings]

    # Import Vault settings source only when explicitly enabled — avoids loading
    # hvac on cold start for apps that ship with the vault extra but leave it off.
    vault_enabled = os.environ.get("VAULT__ENABLED", "").strip().lower() in {
        "1",
        "true",
        "yes",
        "on",
    }
    if vault_enabled:
        try:
            vault_module = importlib.import_module("archipy.configs.vault_settings_source")
        except ImportError as e:
            raise ConfigurationError(
                operation="vault_import",
                reason='hvac is not installed; install with uv add "archipy[vault]"',
            ) from e
        sources.append(vault_module.VaultSettingsSource(settings_cls))

    sources.extend(
        [
            PyprojectTomlConfigSettingsSource(settings_cls),
            TomlConfigSettingsSource(settings_cls),
            env_settings,
            dotenv_settings,
            init_settings,
        ],
    )
    return tuple(sources)

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. It is called automatically by set_global().

When APP_NAME is set, unset or placeholder nested identity fields are filled: OTEL.SERVICE_NAME, FASTAPI.PROJECT_NAME, AUTH.JWT_ISSUER, and TEMPORAL.CLIENT_IDENTITY. Explicit nested values always win.

Examples:

>>> class MyAppConfig(BaseConfig):
...     def customize(self) -> None:
...         super().customize()
...         self.REDIS.MASTER_HOST = "redis.internal"
>>> config = MyAppConfig()
>>> BaseConfig.set_global(config)  # calls customize() automatically
Source code in archipy/configs/base_config.py 
222
223
224
225
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
def customize(self) -> None:
    """Customize configuration after loading.

    This method can be overridden in subclasses to perform
    custom configuration modifications after loading settings.
    It is called automatically by `set_global()`.

    When ``APP_NAME`` is set, unset or placeholder nested identity fields are
    filled: ``OTEL.SERVICE_NAME``, ``FASTAPI.PROJECT_NAME``, ``AUTH.JWT_ISSUER``,
    and ``TEMPORAL.CLIENT_IDENTITY``. Explicit nested values always win.

    Examples:
        >>> class MyAppConfig(BaseConfig):
        ...     def customize(self) -> None:
        ...         super().customize()
        ...         self.REDIS.MASTER_HOST = "redis.internal"
        >>> config = MyAppConfig()
        >>> BaseConfig.set_global(config)  # calls customize() automatically
    """
    if self.OTEL.ENVIRONMENT is None:
        self.OTEL.ENVIRONMENT = self.ENVIRONMENT

    if self.APP_NAME:
        if self.OTEL.SERVICE_NAME is None:
            self.OTEL.SERVICE_NAME = self.APP_NAME
        if self.FASTAPI.PROJECT_NAME == "project_name":
            self.FASTAPI.PROJECT_NAME = self.APP_NAME
        if self.AUTH.JWT_ISSUER == "your-app-name":
            self.AUTH.JWT_ISSUER = self.APP_NAME
        if self.TEMPORAL.CLIENT_IDENTITY is None:
            self.TEMPORAL.CLIENT_IDENTITY = self.APP_NAME

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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
@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
    # Pydantic wraps unset private attrs in ModelPrivateAttr; ``is None`` alone never fires.
    if global_config is None or isinstance(global_config, ModelPrivateAttr):
        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 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
@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 
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
721
722
723
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 
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
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)

            if parsed.scheme and parsed.scheme != self.DRIVER_NAME:
                self.DRIVER_NAME = parsed.scheme

            _extract_postgres_auth_from_url(parsed, self)
            _extract_postgres_host_port_from_url(parsed, self)

            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 
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
@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 
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
@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)

        if parsed.scheme and parsed.scheme != self.DRIVER_NAME:
            self.DRIVER_NAME = parsed.scheme

        _extract_postgres_auth_from_url(parsed, self)
        _extract_postgres_host_port_from_url(parsed, self)

        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 MySQL SQLAlchemy ORM.

Extends SQLAlchemyConfig with MySQL-specific defaults.

Source code in archipy/configs/config_template.py 
827
828
829
830
831
832
833
834
class MySQLSQLAlchemyConfig(SQLAlchemyConfig):
    """Configuration settings for MySQL SQLAlchemy ORM.

    Extends SQLAlchemyConfig with MySQL-specific defaults.
    """

    DRIVER_NAME: str = Field(default="mysql+pymysql", description="MySQL driver name")
    PORT: int | None = Field(default=3306, description="MySQL port")

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

DRIVER_NAME: str = Field(
    default="mysql+pymysql", description="MySQL driver name"
)

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

PORT: int | None = Field(
    default=3306, description="MySQL port"
)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

archipy.configs.config_template.MySQLSQLAlchemyConfig.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.MySQLSQLAlchemyConfig.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.MySQLSQLAlchemyConfig.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.MySQLSQLAlchemyConfig.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.MySQLSQLAlchemyConfig.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.MySQLSQLAlchemyConfig.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.MySQLSQLAlchemyConfig.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.MySQLSQLAlchemyConfig.QUERY_CACHE_SIZE class-attribute instance-attribute 

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

archipy.configs.config_template.MySQLSQLAlchemyConfig.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 SQLite SQLAlchemy ORM.

Extends SQLAlchemyConfig with SQLite-specific settings.

Source code in archipy/configs/config_template.py 
726
727
728
729
730
731
732
733
734
735
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 
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
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 ConfigurationError(
                operation="starrocks",
                reason="isolation_level_not_read_committed",
                additional_data={"got": v},
            )
        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 
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
@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 ConfigurationError(
            operation="starrocks",
            reason="isolation_level_not_read_committed",
            additional_data={"got": v},
        )
    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 
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
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="NetworkTopologyStrategy",
        description="Replication strategy for keyspace creation",
    )
    REPLICATION_CONFIG: dict[str, int] | None = Field(
        default={"datacenter1": 1},
        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="NetworkTopologyStrategy",
    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={"datacenter1": 1},
    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 
1613
1614
1615
1616
1617
1618
1619
1620
1621
@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 
1623
1624
1625
1626
1627
1628
1629
1630
1631
@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 
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
@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 
1648
1649
1650
1651
1652
1653
1654
1655
1656
@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 
 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
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])
            and 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 
100
101
102
103
104
105
106
107
@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 
109
110
111
112
113
114
115
116
117
118
@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])
        and 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

Service Configs

Bases: BaseModel

Configuration settings for Redis cache integration.

Supports standalone, sentinel, and cluster deployments.

Source code in archipy/configs/config_template.py 
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
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")
    SENTINEL_PASSWORD: str | None = Field(default=None, description="Password for sentinel nodes (if distinct)")

    # 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")
    PROTOCOL: Literal[2, 3] = Field(default=3, description="RESP protocol version (2 or 3)")
    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 ConfigurationError(operation="redis", reason="cluster_nodes_required")
            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 ConfigurationError(operation="redis", reason="sentinel_config_incomplete")

        elif self.MODE == RedisMode.STANDALONE:
            if not self.MASTER_HOST:
                raise ConfigurationError(operation="redis", reason="master_host_required")

        return self

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

MODE: RedisMode = Field(
    default=RedisMode.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.SENTINEL_PASSWORD class-attribute instance-attribute 

SENTINEL_PASSWORD: str | None = Field(
    default=None,
    description="Password for sentinel nodes (if distinct)",
)

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.PROTOCOL class-attribute instance-attribute 

PROTOCOL: Literal[2, 3] = Field(
    default=3, description="RESP protocol version (2 or 3)"
)

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 
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
@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 ConfigurationError(operation="redis", reason="cluster_nodes_required")
        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 ConfigurationError(operation="redis", reason="sentinel_config_incomplete")

    elif self.MODE == RedisMode.STANDALONE:
        if not self.MASTER_HOST:
            raise ConfigurationError(operation="redis", reason="master_host_required")

    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 
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
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
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
413
414
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
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
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"] and not (
            self.SASL_MECHANISM and self.USERNAME and self.PASSWORD
        ):
            raise ConfigurationError(operation="kafka_validate", reason="sasl_auth_incomplete")
        if self.SECURITY_PROTOCOL == "SSL" and 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 ConfigurationError(operation="kafka_consumer", reason="auto_commit_offset_reset_conflict")
        if self.HEARTBEAT_INTERVAL_MS >= self.SESSION_TIMEOUT_MS:
            raise ConfigurationError(operation="kafka_consumer", reason="heartbeat_exceeds_session_timeout")
        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 ConfigurationError(operation="kafka_producer", reason="idempotence_requires_acks_all")
        if self.TRANSACTIONAL_ID is not None and not self.ENABLE_IDEMPOTENCE:
            raise ConfigurationError(operation="kafka_producer", reason="transactional_id_requires_idempotence")
        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 
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
@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"] and not (
        self.SASL_MECHANISM and self.USERNAME and self.PASSWORD
    ):
        raise ConfigurationError(operation="kafka_validate", reason="sasl_auth_incomplete")
    if self.SECURITY_PROTOCOL == "SSL" and 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 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
@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 ConfigurationError(operation="kafka_consumer", reason="auto_commit_offset_reset_conflict")
    if self.HEARTBEAT_INTERVAL_MS >= self.SESSION_TIMEOUT_MS:
        raise ConfigurationError(operation="kafka_consumer", reason="heartbeat_exceeds_session_timeout")
    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 
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
@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 ConfigurationError(operation="kafka_producer", reason="idempotence_requires_acks_all")
    if self.TRANSACTIONAL_ID is not None and not self.ENABLE_IDEMPOTENCE:
        raise ConfigurationError(operation="kafka_producer", reason="transactional_id_requires_idempotence")
    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 
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
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 
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
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 
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
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

Bases: BaseModel

Configuration settings for HashiCorp Vault integration.

Controls connection, authentication, TLS, and secret-retrieval parameters for both the BaseConfig settings source and VaultAdapter.

Source code in archipy/configs/config_template.py 
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
536
537
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
583
584
585
586
587
588
589
590
591
592
593
594
595
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
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
class VaultConfig(BaseModel):
    """Configuration settings for HashiCorp Vault integration.

    Controls connection, authentication, TLS, and secret-retrieval parameters
    for both the BaseConfig settings source and VaultAdapter.
    """

    ENABLED: bool = Field(default=False, description="Whether Vault integration is active")
    ADDR: str | None = Field(default=None, description="Vault server address, e.g. https://vault.example.com:8200")
    NAMESPACE: str | None = Field(default=None, description="Vault Enterprise namespace")
    AUTH_METHOD: Literal[
        "token",
        "approle",
        "kubernetes",
        "userpass",
        "ldap",
        "okta",
        "jwt",
        "aws",
        "azure",
        "gcp",
        "github",
        "cert",
    ] = Field(
        default="token",
        description="Vault auth method (hvac-supported backends)",
    )

    # Token auth
    TOKEN: str | None = Field(default=None, description="Vault token (token auth)")
    TOKEN_FILE: str | None = Field(
        default=None,
        description="Path to a file containing the Vault token; takes precedence over TOKEN",
    )

    # AppRole auth
    APPROLE_ROLE_ID: str | None = Field(default=None, description="AppRole role_id")
    APPROLE_SECRET_ID: str | None = Field(default=None, description="AppRole secret_id")
    APPROLE_SECRET_ID_FILE: str | None = Field(
        default=None,
        description="Path to a file containing the AppRole secret_id; takes precedence over APPROLE_SECRET_ID",
    )
    APPROLE_MOUNT_POINT: str = Field(default="approle", description="Mount point of the AppRole auth method")

    # Kubernetes auth
    KUBERNETES_ROLE: str | None = Field(
        default=None,
        description="Vault role bound to the Kubernetes service account",
    )
    KUBERNETES_JWT_PATH: str = Field(
        default="/var/run/secrets/kubernetes.io/serviceaccount/token",
        description="Path to the service account JWT used for Kubernetes auth",
    )
    KUBERNETES_MOUNT_POINT: str = Field(
        default="kubernetes",
        description="Mount point of the Kubernetes auth method",
    )

    # Username/password auth (userpass, ldap, okta)
    USERNAME: str | None = Field(default=None, description="Username for userpass/ldap/okta auth")
    PASSWORD: str | None = Field(default=None, description="Password for userpass/ldap/okta auth")
    PASSWORD_FILE: str | None = Field(
        default=None,
        description="Path to a file containing the password; takes precedence over PASSWORD",
    )
    USERPASS_MOUNT_POINT: str = Field(default="userpass", description="Mount point of the userpass auth method")
    LDAP_MOUNT_POINT: str = Field(default="ldap", description="Mount point of the LDAP auth method")
    OKTA_MOUNT_POINT: str = Field(default="okta", description="Mount point of the Okta auth method")

    # JWT auth
    JWT_ROLE: str | None = Field(default=None, description="Role name for JWT auth")
    JWT: str | None = Field(default=None, description="JWT for JWT auth")
    JWT_FILE: str | None = Field(
        default=None,
        description="Path to a file containing the JWT; takes precedence over JWT",
    )
    JWT_MOUNT_POINT: str = Field(default="jwt", description="Mount path of the JWT auth method")

    # AWS IAM auth
    AWS_ACCESS_KEY: str | None = Field(default=None, description="AWS access key for IAM auth")
    AWS_SECRET_KEY: str | None = Field(default=None, description="AWS secret key for IAM auth")
    AWS_SESSION_TOKEN: str | None = Field(default=None, description="Optional AWS session token for IAM auth")
    AWS_ROLE: str | None = Field(default=None, description="Optional Vault role for AWS IAM auth")
    AWS_REGION: str = Field(default="us-east-1", description="AWS region used for IAM auth signing")
    AWS_MOUNT_POINT: str = Field(default="aws", description="Mount point of the AWS auth method")

    # Azure auth
    AZURE_ROLE: str | None = Field(default=None, description="Role name for Azure auth")
    AZURE_JWT: str | None = Field(default=None, description="JWT for Azure auth")
    AZURE_JWT_FILE: str | None = Field(
        default=None,
        description="Path to a file containing the Azure JWT; takes precedence over AZURE_JWT",
    )
    AZURE_SUBSCRIPTION_ID: str | None = Field(default=None, description="Optional Azure subscription id")
    AZURE_RESOURCE_GROUP_NAME: str | None = Field(default=None, description="Optional Azure resource group name")
    AZURE_VM_NAME: str | None = Field(default=None, description="Optional Azure VM name")
    AZURE_VMSS_NAME: str | None = Field(default=None, description="Optional Azure VMSS name")
    AZURE_MOUNT_POINT: str = Field(default="azure", description="Mount point of the Azure auth method")

    # GCP auth
    GCP_ROLE: str | None = Field(default=None, description="Role name for GCP auth")
    GCP_JWT: str | None = Field(default=None, description="JWT for GCP auth")
    GCP_JWT_FILE: str | None = Field(
        default=None,
        description="Path to a file containing the GCP JWT; takes precedence over GCP_JWT",
    )
    GCP_MOUNT_POINT: str = Field(default="gcp", description="Mount point of the GCP auth method")

    # GitHub auth
    GITHUB_TOKEN: str | None = Field(default=None, description="GitHub personal access token for GitHub auth")
    GITHUB_TOKEN_FILE: str | None = Field(
        default=None,
        description="Path to a file containing the GitHub token; takes precedence over GITHUB_TOKEN",
    )
    GITHUB_MOUNT_POINT: str = Field(default="github", description="Mount point of the GitHub auth method")

    # Cert auth (TLS client certificate auth method; often paired with CLIENT_CERT_PATH/KEY)
    CERT_NAME: str = Field(
        default="",
        description="Optional named cert role for the cert auth method",
    )
    CERT_MOUNT_POINT: str = Field(default="cert", description="Mount point of the cert auth method")

    # KV v2
    MOUNT_POINT: str = Field(default="secret", description="Mount point of the KV v2 secrets engine")
    SECRET_PATHS: list[str] = Field(
        default=[],
        description="KV v2 paths read by the settings source; later paths win",
    )

    # TLS / mTLS
    VERIFY_SSL: bool = Field(default=True, description="Verify the Vault server's TLS certificate")
    CA_CERT_PATH: str | None = Field(
        default=None,
        description="Path to a CA bundle for verifying Vault's TLS certificate",
    )
    CLIENT_CERT_PATH: str | None = Field(
        default=None,
        description="Path to a client certificate for mutual TLS",
    )
    CLIENT_KEY_PATH: str | None = Field(
        default=None,
        description="Path to the client private key for mutual TLS",
    )

    # Timeouts / retries
    CONNECT_TIMEOUT: float = Field(default=5.0, description="Connection timeout in seconds")
    READ_TIMEOUT: float = Field(default=10.0, description="Read timeout in seconds")
    RETRIES_MAX_ATTEMPTS: int = Field(default=3, description="Maximum retry attempts for failed requests")

    # Token lifecycle (adapter only — the settings source is one-shot at startup)
    AUTO_RENEW_TOKEN: bool = Field(
        default=False,
        description="Automatically renew the token before it expires",
    )
    RENEW_THRESHOLD_SECONDS: int = Field(
        default=60,
        description="Renew the token once its remaining TTL drops below this many seconds",
    )

    # Secret caching (adapter only)
    SECRET_CACHE_TTL: int = Field(
        default=0,
        description="Cache VaultAdapter.read_secret() results for this many seconds; 0 disables caching",
    )

archipy.configs.config_template.VaultConfig.ENABLED class-attribute instance-attribute 

ENABLED: bool = Field(
    default=False,
    description="Whether Vault integration is active",
)

archipy.configs.config_template.VaultConfig.ADDR class-attribute instance-attribute 

ADDR: str | None = Field(
    default=None,
    description="Vault server address, e.g. https://vault.example.com:8200",
)

archipy.configs.config_template.VaultConfig.NAMESPACE class-attribute instance-attribute 

NAMESPACE: str | None = Field(
    default=None, description="Vault Enterprise namespace"
)

archipy.configs.config_template.VaultConfig.AUTH_METHOD class-attribute instance-attribute 

AUTH_METHOD: Literal[
    "token",
    "approle",
    "kubernetes",
    "userpass",
    "ldap",
    "okta",
    "jwt",
    "aws",
    "azure",
    "gcp",
    "github",
    "cert",
] = Field(
    default="token",
    description="Vault auth method (hvac-supported backends)",
)

archipy.configs.config_template.VaultConfig.TOKEN class-attribute instance-attribute 

TOKEN: str | None = Field(
    default=None, description="Vault token (token auth)"
)

archipy.configs.config_template.VaultConfig.TOKEN_FILE class-attribute instance-attribute 

TOKEN_FILE: str | None = Field(
    default=None,
    description="Path to a file containing the Vault token; takes precedence over TOKEN",
)

archipy.configs.config_template.VaultConfig.APPROLE_ROLE_ID class-attribute instance-attribute 

APPROLE_ROLE_ID: str | None = Field(
    default=None, description="AppRole role_id"
)

archipy.configs.config_template.VaultConfig.APPROLE_SECRET_ID class-attribute instance-attribute 

APPROLE_SECRET_ID: str | None = Field(
    default=None, description="AppRole secret_id"
)

archipy.configs.config_template.VaultConfig.APPROLE_SECRET_ID_FILE class-attribute instance-attribute 

APPROLE_SECRET_ID_FILE: str | None = Field(
    default=None,
    description="Path to a file containing the AppRole secret_id; takes precedence over APPROLE_SECRET_ID",
)

archipy.configs.config_template.VaultConfig.APPROLE_MOUNT_POINT class-attribute instance-attribute 

APPROLE_MOUNT_POINT: str = Field(
    default="approle",
    description="Mount point of the AppRole auth method",
)

archipy.configs.config_template.VaultConfig.KUBERNETES_ROLE class-attribute instance-attribute 

KUBERNETES_ROLE: str | None = Field(
    default=None,
    description="Vault role bound to the Kubernetes service account",
)

archipy.configs.config_template.VaultConfig.KUBERNETES_JWT_PATH class-attribute instance-attribute 

KUBERNETES_JWT_PATH: str = Field(
    default="/var/run/secrets/kubernetes.io/serviceaccount/token",
    description="Path to the service account JWT used for Kubernetes auth",
)

archipy.configs.config_template.VaultConfig.KUBERNETES_MOUNT_POINT class-attribute instance-attribute 

KUBERNETES_MOUNT_POINT: str = Field(
    default="kubernetes",
    description="Mount point of the Kubernetes auth method",
)

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

USERNAME: str | None = Field(
    default=None,
    description="Username for userpass/ldap/okta auth",
)

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

PASSWORD: str | None = Field(
    default=None,
    description="Password for userpass/ldap/okta auth",
)

archipy.configs.config_template.VaultConfig.PASSWORD_FILE class-attribute instance-attribute 

PASSWORD_FILE: str | None = Field(
    default=None,
    description="Path to a file containing the password; takes precedence over PASSWORD",
)

archipy.configs.config_template.VaultConfig.USERPASS_MOUNT_POINT class-attribute instance-attribute 

USERPASS_MOUNT_POINT: str = Field(
    default="userpass",
    description="Mount point of the userpass auth method",
)

archipy.configs.config_template.VaultConfig.LDAP_MOUNT_POINT class-attribute instance-attribute 

LDAP_MOUNT_POINT: str = Field(
    default="ldap",
    description="Mount point of the LDAP auth method",
)

archipy.configs.config_template.VaultConfig.OKTA_MOUNT_POINT class-attribute instance-attribute 

OKTA_MOUNT_POINT: str = Field(
    default="okta",
    description="Mount point of the Okta auth method",
)

archipy.configs.config_template.VaultConfig.JWT_ROLE class-attribute instance-attribute 

JWT_ROLE: str | None = Field(
    default=None, description="Role name for JWT auth"
)

archipy.configs.config_template.VaultConfig.JWT class-attribute instance-attribute 

JWT: str | None = Field(
    default=None, description="JWT for JWT auth"
)

archipy.configs.config_template.VaultConfig.JWT_FILE class-attribute instance-attribute 

JWT_FILE: str | None = Field(
    default=None,
    description="Path to a file containing the JWT; takes precedence over JWT",
)

archipy.configs.config_template.VaultConfig.JWT_MOUNT_POINT class-attribute instance-attribute 

JWT_MOUNT_POINT: str = Field(
    default="jwt",
    description="Mount path of the JWT auth method",
)

archipy.configs.config_template.VaultConfig.AWS_ACCESS_KEY class-attribute instance-attribute 

AWS_ACCESS_KEY: str | None = Field(
    default=None, description="AWS access key for IAM auth"
)

archipy.configs.config_template.VaultConfig.AWS_SECRET_KEY class-attribute instance-attribute 

AWS_SECRET_KEY: str | None = Field(
    default=None, description="AWS secret key for IAM auth"
)

archipy.configs.config_template.VaultConfig.AWS_SESSION_TOKEN class-attribute instance-attribute 

AWS_SESSION_TOKEN: str | None = Field(
    default=None,
    description="Optional AWS session token for IAM auth",
)

archipy.configs.config_template.VaultConfig.AWS_ROLE class-attribute instance-attribute 

AWS_ROLE: str | None = Field(
    default=None,
    description="Optional Vault role for AWS IAM auth",
)

archipy.configs.config_template.VaultConfig.AWS_REGION class-attribute instance-attribute 

AWS_REGION: str = Field(
    default="us-east-1",
    description="AWS region used for IAM auth signing",
)

archipy.configs.config_template.VaultConfig.AWS_MOUNT_POINT class-attribute instance-attribute 

AWS_MOUNT_POINT: str = Field(
    default="aws",
    description="Mount point of the AWS auth method",
)

archipy.configs.config_template.VaultConfig.AZURE_ROLE class-attribute instance-attribute 

AZURE_ROLE: str | None = Field(
    default=None, description="Role name for Azure auth"
)

archipy.configs.config_template.VaultConfig.AZURE_JWT class-attribute instance-attribute 

AZURE_JWT: str | None = Field(
    default=None, description="JWT for Azure auth"
)

archipy.configs.config_template.VaultConfig.AZURE_JWT_FILE class-attribute instance-attribute 

AZURE_JWT_FILE: str | None = Field(
    default=None,
    description="Path to a file containing the Azure JWT; takes precedence over AZURE_JWT",
)

archipy.configs.config_template.VaultConfig.AZURE_SUBSCRIPTION_ID class-attribute instance-attribute 

AZURE_SUBSCRIPTION_ID: str | None = Field(
    default=None,
    description="Optional Azure subscription id",
)

archipy.configs.config_template.VaultConfig.AZURE_RESOURCE_GROUP_NAME class-attribute instance-attribute 

AZURE_RESOURCE_GROUP_NAME: str | None = Field(
    default=None,
    description="Optional Azure resource group name",
)

archipy.configs.config_template.VaultConfig.AZURE_VM_NAME class-attribute instance-attribute 

AZURE_VM_NAME: str | None = Field(
    default=None, description="Optional Azure VM name"
)

archipy.configs.config_template.VaultConfig.AZURE_VMSS_NAME class-attribute instance-attribute 

AZURE_VMSS_NAME: str | None = Field(
    default=None, description="Optional Azure VMSS name"
)

archipy.configs.config_template.VaultConfig.AZURE_MOUNT_POINT class-attribute instance-attribute 

AZURE_MOUNT_POINT: str = Field(
    default="azure",
    description="Mount point of the Azure auth method",
)

archipy.configs.config_template.VaultConfig.GCP_ROLE class-attribute instance-attribute 

GCP_ROLE: str | None = Field(
    default=None, description="Role name for GCP auth"
)

archipy.configs.config_template.VaultConfig.GCP_JWT class-attribute instance-attribute 

GCP_JWT: str | None = Field(
    default=None, description="JWT for GCP auth"
)

archipy.configs.config_template.VaultConfig.GCP_JWT_FILE class-attribute instance-attribute 

GCP_JWT_FILE: str | None = Field(
    default=None,
    description="Path to a file containing the GCP JWT; takes precedence over GCP_JWT",
)

archipy.configs.config_template.VaultConfig.GCP_MOUNT_POINT class-attribute instance-attribute 

GCP_MOUNT_POINT: str = Field(
    default="gcp",
    description="Mount point of the GCP auth method",
)

archipy.configs.config_template.VaultConfig.GITHUB_TOKEN class-attribute instance-attribute 

GITHUB_TOKEN: str | None = Field(
    default=None,
    description="GitHub personal access token for GitHub auth",
)

archipy.configs.config_template.VaultConfig.GITHUB_TOKEN_FILE class-attribute instance-attribute 

GITHUB_TOKEN_FILE: str | None = Field(
    default=None,
    description="Path to a file containing the GitHub token; takes precedence over GITHUB_TOKEN",
)

archipy.configs.config_template.VaultConfig.GITHUB_MOUNT_POINT class-attribute instance-attribute 

GITHUB_MOUNT_POINT: str = Field(
    default="github",
    description="Mount point of the GitHub auth method",
)

archipy.configs.config_template.VaultConfig.CERT_NAME class-attribute instance-attribute 

CERT_NAME: str = Field(
    default="",
    description="Optional named cert role for the cert auth method",
)

archipy.configs.config_template.VaultConfig.CERT_MOUNT_POINT class-attribute instance-attribute 

CERT_MOUNT_POINT: str = Field(
    default="cert",
    description="Mount point of the cert auth method",
)

archipy.configs.config_template.VaultConfig.MOUNT_POINT class-attribute instance-attribute 

MOUNT_POINT: str = Field(
    default="secret",
    description="Mount point of the KV v2 secrets engine",
)

archipy.configs.config_template.VaultConfig.SECRET_PATHS class-attribute instance-attribute 

SECRET_PATHS: list[str] = Field(
    default=[],
    description="KV v2 paths read by the settings source; later paths win",
)

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

VERIFY_SSL: bool = Field(
    default=True,
    description="Verify the Vault server's TLS certificate",
)

archipy.configs.config_template.VaultConfig.CA_CERT_PATH class-attribute instance-attribute 

CA_CERT_PATH: str | None = Field(
    default=None,
    description="Path to a CA bundle for verifying Vault's TLS certificate",
)

archipy.configs.config_template.VaultConfig.CLIENT_CERT_PATH class-attribute instance-attribute 

CLIENT_CERT_PATH: str | None = Field(
    default=None,
    description="Path to a client certificate for mutual TLS",
)

archipy.configs.config_template.VaultConfig.CLIENT_KEY_PATH class-attribute instance-attribute 

CLIENT_KEY_PATH: str | None = Field(
    default=None,
    description="Path to the client private key for mutual TLS",
)

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

CONNECT_TIMEOUT: float = Field(
    default=5.0, description="Connection timeout in seconds"
)

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

READ_TIMEOUT: float = Field(
    default=10.0, description="Read timeout in seconds"
)

archipy.configs.config_template.VaultConfig.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.VaultConfig.AUTO_RENEW_TOKEN class-attribute instance-attribute 

AUTO_RENEW_TOKEN: bool = Field(
    default=False,
    description="Automatically renew the token before it expires",
)

archipy.configs.config_template.VaultConfig.RENEW_THRESHOLD_SECONDS class-attribute instance-attribute 

RENEW_THRESHOLD_SECONDS: int = Field(
    default=60,
    description="Renew the token once its remaining TTL drops below this many seconds",
)

archipy.configs.config_template.VaultConfig.SECRET_CACHE_TTL class-attribute instance-attribute 

SECRET_CACHE_TTL: int = Field(
    default=0,
    description="Cache VaultAdapter.read_secret() results for this many seconds; 0 disables caching",
)

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
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
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_ORIGIN_REGEX: str | None = Field(
        default=None,
        description="Regex pattern for allowed origins",
    )
    CORS_MIDDLEWARE_ALLOW_ORIGINS: list[str] = Field(default=["*"], description="Allowed CORS origins")
    CORS_MIDDLEWARE_EXPOSE_HEADERS: list[str] = Field(default=[], description="Exposed CORS headers")
    CORS_MIDDLEWARE_MAX_AGE: int = Field(default=600, description="Preflight cache duration in seconds")
    GZIP_MIDDLEWARE_IS_ENABLED: bool = Field(default=False, description="Whether GZip response compression is enabled")
    GZIP_MIDDLEWARE_MINIMUM_SIZE: int = Field(
        default=500,
        ge=0,
        description="Minimum response body size in bytes before compression is applied",
    )
    GZIP_MIDDLEWARE_COMPRESSLEVEL: int = Field(
        default=6,
        ge=1,
        le=9,
        description="GZip compression level (1=fastest, 9=best compression)",
    )
    TRUSTED_HOST_MIDDLEWARE_IS_ENABLED: bool = Field(
        default=False,
        description="Whether TrustedHost middleware validates the Host header",
    )
    TRUSTED_HOST_MIDDLEWARE_ALLOWED_HOSTS: list[str] = Field(
        default=[],
        description="Allowed hostnames (e.g. example.com, *.example.com)",
    )
    TRUSTED_HOST_MIDDLEWARE_WWW_REDIRECT: bool = Field(
        default=True,
        description="Redirect www hosts to bare domain when allowed",
    )
    HTTPS_REDIRECT_MIDDLEWARE_IS_ENABLED: bool = Field(
        default=False,
        description="Whether HTTP requests are redirected to HTTPS",
    )
    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="127.0.0.1",
        description="Host to serve the application on (set to 0.0.0.0 for container/public binding)",
    )
    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_ORIGIN_REGEX class-attribute instance-attribute 

CORS_MIDDLEWARE_ALLOW_ORIGIN_REGEX: str | None = Field(
    default=None,
    description="Regex pattern for allowed origins",
)

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.CORS_MIDDLEWARE_EXPOSE_HEADERS class-attribute instance-attribute 

CORS_MIDDLEWARE_EXPOSE_HEADERS: list[str] = Field(
    default=[], description="Exposed CORS headers"
)

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

CORS_MIDDLEWARE_MAX_AGE: int = Field(
    default=600,
    description="Preflight cache duration in seconds",
)

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

GZIP_MIDDLEWARE_IS_ENABLED: bool = Field(
    default=False,
    description="Whether GZip response compression is enabled",
)

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

GZIP_MIDDLEWARE_MINIMUM_SIZE: int = Field(
    default=500,
    ge=0,
    description="Minimum response body size in bytes before compression is applied",
)

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

GZIP_MIDDLEWARE_COMPRESSLEVEL: int = Field(
    default=6,
    ge=1,
    le=9,
    description="GZip compression level (1=fastest, 9=best compression)",
)

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

TRUSTED_HOST_MIDDLEWARE_IS_ENABLED: bool = Field(
    default=False,
    description="Whether TrustedHost middleware validates the Host header",
)

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

TRUSTED_HOST_MIDDLEWARE_ALLOWED_HOSTS: list[str] = Field(
    default=[],
    description="Allowed hostnames (e.g. example.com, *.example.com)",
)

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

TRUSTED_HOST_MIDDLEWARE_WWW_REDIRECT: bool = Field(
    default=True,
    description="Redirect www hosts to bare domain when allowed",
)

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

HTTPS_REDIRECT_MIDDLEWARE_IS_ENABLED: bool = Field(
    default=False,
    description="Whether HTTP requests are redirected to HTTPS",
)

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="127.0.0.1",
    description="Host to serve the application on (set to 0.0.0.0 for container/public binding)",
)

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 for gRPC server rate limiting.

Controls Redis key layout, failure modes, and identity resolution for GrpcServerRateLimitInterceptor / AsyncGrpcServerRateLimitInterceptor. Per-RPC limits are declared via grpc_rate_limit_decorator on servicer methods. When IS_ENABLED is True, AppUtils.create_grpc_app and AppUtils.create_async_grpc_app register the matching interceptor automatically.

Source code in archipy/configs/config_template.py 
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
224
225
226
227
228
229
230
231
232
233
234
class GrpcRateLimitConfig(BaseModel):
    """Configuration for gRPC server rate limiting.

    Controls Redis key layout, failure modes, and identity resolution for
    ``GrpcServerRateLimitInterceptor`` / ``AsyncGrpcServerRateLimitInterceptor``.
    Per-RPC limits are declared via ``grpc_rate_limit_decorator`` on servicer methods.
    When ``IS_ENABLED`` is True, ``AppUtils.create_grpc_app`` and ``AppUtils.create_async_grpc_app``
    register the matching interceptor automatically.
    """

    IS_ENABLED: bool = Field(
        default=False,
        description=(
            "When True, register the gRPC rate-limit interceptor in "
            "``AppUtils.create_grpc_app`` / ``AppUtils.create_async_grpc_app``."
        ),
    )
    KEY_PREFIX: str | None = Field(
        default=None,
        description="Redis key prefix. When None, ``{GRPC.SERVE_HOST}:RateLimit`` is used.",
    )
    FAIL_CLOSED: bool = Field(
        default=True,
        description="When True, Redis or identity-resolution failures abort with UNAVAILABLE instead of allowing RPCs.",
    )
    SKIP_METHODS: list[str] = Field(
        default=["/grpc.health.v1.Health/Check", "/grpc.health.v1.Health/Watch"],
        description="Full gRPC method names excluded from rate limiting.",
    )
    IDENTITY_FROM_ACCESS_TOKEN: bool = Field(
        default=True,
        description=(
            "When True, verify Bearer access tokens from invocation metadata via JWTUtils and bucket by sub; "
            "fall back to peer IP when the token is missing or invalid."
        ),
    )

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

IS_ENABLED: bool = Field(
    default=False,
    description="When True, register the gRPC rate-limit interceptor in ``AppUtils.create_grpc_app`` / ``AppUtils.create_async_grpc_app``.",
)

archipy.configs.config_template.GrpcRateLimitConfig.KEY_PREFIX class-attribute instance-attribute 

KEY_PREFIX: str | None = Field(
    default=None,
    description="Redis key prefix. When None, ``{GRPC.SERVE_HOST}:RateLimit`` is used.",
)

archipy.configs.config_template.GrpcRateLimitConfig.FAIL_CLOSED class-attribute instance-attribute 

FAIL_CLOSED: bool = Field(
    default=True,
    description="When True, Redis or identity-resolution failures abort with UNAVAILABLE instead of allowing RPCs.",
)

archipy.configs.config_template.GrpcRateLimitConfig.SKIP_METHODS class-attribute instance-attribute 

SKIP_METHODS: list[str] = Field(
    default=[
        "/grpc.health.v1.Health/Check",
        "/grpc.health.v1.Health/Watch",
    ],
    description="Full gRPC method names excluded from rate limiting.",
)

archipy.configs.config_template.GrpcRateLimitConfig.IDENTITY_FROM_ACCESS_TOKEN class-attribute instance-attribute 

IDENTITY_FROM_ACCESS_TOKEN: bool = Field(
    default=True,
    description="When True, verify Bearer access tokens from invocation metadata via JWTUtils and bucket by sub; fall back to peer IP when the token is missing or invalid.",
)

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 
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
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. If omitted (None), set at model validation to "
            "THREAD_PER_CPU_CORE * (os.cpu_count() or 1)."
        ),
    )
    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=[
            # ── Message limits ──────────────────────────────────────────────
            ("grpc.max_metadata_size", 16 * 1024),  # 16KB  (default is 8KB)
            ("grpc.max_message_length", 128 * 1024 * 1024),  # 128MB
            ("grpc.max_receive_message_length", 128 * 1024 * 1024),
            ("grpc.max_send_message_length", 128 * 1024 * 1024),
            # ── Keepalive ───────────────────────────────────────────────────
            # Server rarely initiates pings; 2h is the standard default.
            ("grpc.keepalive_time_ms", 7_200_000),  # 2h
            ("grpc.keepalive_timeout_ms", 20_000),  # 20s
            ("grpc.keepalive_permit_without_calls", 1),  # allow pings on idle connections
            # ── Ping enforcement (server-side) ──────────────────────────────
            # Must be <= client keepalive_time_ms to avoid ENHANCE_YOUR_CALM.
            ("grpc.http2.min_recv_ping_interval_without_data_ms", 300_000),  # 5min
            ("grpc.http2.min_ping_interval_without_data_ms", 300_000),  # 5min
            ("grpc.http2.max_pings_without_data", 0),
            ("grpc.http2.max_ping_strikes", 2),
            # ── Connection lifetime ─────────────────────────────────────────
            ("grpc.max_connection_idle_ms", 600_000),  # 10min
            ("grpc.max_connection_age_ms", 1_800_000),  # 30min
            ("grpc.max_connection_age_grace_ms", 5_000),  # 5s graceful drain
        ],
        description="Server channel options",
    )

    STUB_OPTIONS_CONFIG_LIST: list[tuple[str, int | str]] = Field(
        default=[
            # ── Message limits ──────────────────────────────────────────────
            ("grpc.max_metadata_size", 16 * 1024),  # 16KB
            ("grpc.max_message_length", 128 * 1024 * 1024),  # 128MB
            ("grpc.max_receive_message_length", 128 * 1024 * 1024),
            ("grpc.max_send_message_length", 128 * 1024 * 1024),
            # ── Keepalive ───────────────────────────────────────────────────
            # 5min is the recommended minimum to avoid ENHANCE_YOUR_CALM / GOAWAY.
            # Also activates TCP_USER_TIMEOUT for fast dead-connection detection.
            ("grpc.keepalive_time_ms", 300_000),  # 5min
            ("grpc.keepalive_timeout_ms", 20_000),  # 20s
            ("grpc.keepalive_permit_without_calls", 1),
            ("grpc.http2.max_pings_without_data", 0),
            (
                "grpc.service_config",
                "{"
                '  "methodConfig": [{'
                '    "name": [],'
                '    "timeout": "10s",'
                '    "waitForReady": true,'
                '    "retryPolicy": {'
                '      "maxAttempts": 5,'
                '      "initialBackoff": "0.1s",'
                '      "maxBackoff": "1s",'
                '      "backoffMultiplier": 2,'
                '      "retryableStatusCodes": ["UNAVAILABLE", "ABORTED", "RESOURCE_EXHAUSTED"]'
                "    }"
                "  }],"
                '  "retryThrottling": {'
                '    "maxTokens": 10,'
                '    "tokenRatio": 0.1'
                "  }"
                "}",
            ),
        ],
        description="Client stub channel options",
    )

    @model_validator(mode="after")
    def resolve_thread_worker_count(self) -> Self:
        """Resolve worker thread count from CPU cores when not explicitly set.

        Returns:
            This model with ``THREAD_WORKER_COUNT`` set when it was ``None``.
        """
        if self.THREAD_WORKER_COUNT is not None:
            return self
        cores = os.cpu_count() or 1
        resolved = self.THREAD_PER_CPU_CORE * cores
        self.THREAD_WORKER_COUNT = resolved
        return self

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. If omitted (None), set at model validation to THREAD_PER_CPU_CORE * (os.cpu_count() or 1).",
)

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. Adjust based on thread block to CPU time ratio.",
)

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", 16 * 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", 7200000),
        ("grpc.keepalive_timeout_ms", 20000),
        ("grpc.keepalive_permit_without_calls", 1),
        (
            "grpc.http2.min_recv_ping_interval_without_data_ms",
            300000,
        ),
        (
            "grpc.http2.min_ping_interval_without_data_ms",
            300000,
        ),
        ("grpc.http2.max_pings_without_data", 0),
        ("grpc.http2.max_ping_strikes", 2),
        ("grpc.max_connection_idle_ms", 600000),
        ("grpc.max_connection_age_ms", 1800000),
        ("grpc.max_connection_age_grace_ms", 5000),
    ],
    description="Server channel 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", 16 * 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", 300000),
            ("grpc.keepalive_timeout_ms", 20000),
            ("grpc.keepalive_permit_without_calls", 1),
            ("grpc.http2.max_pings_without_data", 0),
            (
                "grpc.service_config",
                '{  "methodConfig": [{    "name": [],    "timeout": "10s",    "waitForReady": true,    "retryPolicy": {      "maxAttempts": 5,      "initialBackoff": "0.1s",      "maxBackoff": "1s",      "backoffMultiplier": 2,      "retryableStatusCodes": ["UNAVAILABLE", "ABORTED", "RESOURCE_EXHAUSTED"]    }  }],  "retryThrottling": {    "maxTokens": 10,    "tokenRatio": 0.1  }}',
            ),
        ],
        description="Client stub channel options",
    )
)

archipy.configs.config_template.GrpcConfig.resolve_thread_worker_count 

resolve_thread_worker_count() -> Self

Resolve worker thread count from CPU cores when not explicitly set.

Returns:

Type Description
Self

This model with THREAD_WORKER_COUNT set when it was None.
Source code in archipy/configs/config_template.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
@model_validator(mode="after")
def resolve_thread_worker_count(self) -> Self:
    """Resolve worker thread count from CPU cores when not explicitly set.

    Returns:
        This model with ``THREAD_WORKER_COUNT`` set when it was ``None``.
    """
    if self.THREAD_WORKER_COUNT is not None:
        return self
    cores = os.cpu_count() or 1
    resolved = self.THREAD_PER_CPU_CORE * cores
    self.THREAD_WORKER_COUNT = resolved
    return self

options: show_root_toc_entry: false heading_level: 3

Observability Configs

Bases: BaseModel

Configuration settings for OpenTelemetry observability.

Controls OTLP export of traces, metrics, and logs. Providers are built programmatically from this config — OTEL_* environment-variable autoconfiguration is not used.

Source code in archipy/configs/config_template.py 
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 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
class OpentelemetryConfig(BaseModel):
    """Configuration settings for OpenTelemetry observability.

    Controls OTLP export of traces, metrics, and logs. Providers are built
    programmatically from this config — ``OTEL_*`` environment-variable
    autoconfiguration is not used.
    """

    IS_ENABLED: bool = Field(default=False, description="Master switch for OpenTelemetry")
    TRACES_ENABLED: bool = Field(default=True, description="Export traces when OTel is enabled")
    METRICS_ENABLED: bool = Field(default=True, description="Export metrics when OTel is enabled")
    LOGS_ENABLED: bool = Field(default=True, description="Export logs when OTel is enabled")
    LOGS_EXPORTER: OtelLogsExporter = Field(
        default=OtelLogsExporter.CONSOLE,
        description=(
            "Unique logs exporter when LOGS_ENABLED is true: "
            "console (stdout for INFO/DEBUG, stderr for WARNING+) or otlp"
        ),
    )
    SERVICE_NAME: str | None = Field(default=None, description="OTel resource service.name")
    OTLP_ENDPOINT: HttpUrl = Field(
        default=HttpUrl("http://localhost:4317"),
        description=(
            "Default OTLP collector endpoint (gRPC port 4317; HTTP/protobuf typically 4318). "
            "For http/protobuf, signal paths (/v1/traces, /v1/metrics, /v1/logs) are appended "
            "when the URL has no path. Override per signal with TRACES/METRICS/LOGS_ENDPOINT."
        ),
    )
    TRACES_ENDPOINT: HttpUrl | None = Field(
        default=None,
        description="Optional per-signal OTLP traces endpoint (overrides OTLP_ENDPOINT)",
    )
    METRICS_ENDPOINT: HttpUrl | None = Field(
        default=None,
        description="Optional per-signal OTLP metrics endpoint (overrides OTLP_ENDPOINT)",
    )
    LOGS_ENDPOINT: HttpUrl | None = Field(
        default=None,
        description="Optional per-signal OTLP logs endpoint (overrides OTLP_ENDPOINT)",
    )
    PROTOCOL: Literal["grpc", "http/protobuf"] = Field(
        default="grpc",
        description="OTLP transport protocol",
    )
    OTLP_HEADERS: dict[str, str] = Field(
        default_factory=dict,
        description="Headers sent with each OTLP export (auth tokens, routing metadata)",
    )
    TIMEOUT: float = Field(default=10.0, ge=0.1, description="OTLP export timeout in seconds")
    TRACES_SAMPLE_RATIO: float = Field(
        default=0.1,
        ge=0.0,
        le=1.0,
        description="Parent-based trace ID ratio sampler (inbound sampled traces always continue)",
    )
    RESOURCE_ATTRIBUTES: dict[str, str] = Field(
        default_factory=dict,
        description="Extra OTel resource attributes merged into the Resource",
    )
    METRIC_EXPORT_INTERVAL_MS: int = Field(
        default=60000,
        ge=1000,
        description="Periodic OTLP metric export interval in milliseconds (used when METRICS_EXPORTER=otlp)",
    )
    METRICS_EXPORTER: OtelMetricsExporter = Field(
        default=OtelMetricsExporter.OTLP,
        description="Unique metrics exporter when METRICS_ENABLED is true (otlp, pull, or pushgateway)",
    )
    METRICS_PULL_HOST: str = Field(
        default="0.0.0.0",  # noqa: S104 — intentional scrape bind-all default; override via env
        description="Bind host for the metrics pull scrape server when METRICS_EXPORTER=pull",
    )
    METRICS_PULL_PORT: int = Field(
        default=8200,
        ge=1,
        le=65535,
        description="Bind port for the metrics pull scrape server (/metrics)",
    )
    METRICS_PUSHGATEWAY_URL: str | None = Field(
        default=None,
        description="Prometheus Pushgateway base URL when METRICS_EXPORTER=pushgateway (e.g. http://host:9091)",
    )
    METRICS_PUSHGATEWAY_JOB: str | None = Field(
        default=None,
        description="Pushgateway job label (defaults to SERVICE_NAME when unset)",
    )
    METRICS_PUSHGATEWAY_INTERVAL_SECONDS: int = Field(
        default=60,
        ge=1,
        description="Interval between Pushgateway pushes when METRICS_EXPORTER=pushgateway",
    )
    METRICS_PUSHGATEWAY_TIMEOUT_SECONDS: float = Field(
        default=10.0,
        gt=0,
        description="HTTP timeout in seconds for Pushgateway push/delete calls",
    )
    METRICS_PUSHGATEWAY_GROUPING_KEY: dict[str, str] = Field(
        default_factory=dict,
        max_length=16,
        description="Extra Pushgateway grouping labels (instance, component, ...)",
    )
    METRICS_PUSHGATEWAY_DELETE_ON_SHUTDOWN: bool = Field(
        default=True,
        description="Delete the Pushgateway grouping key on graceful OTel shutdown",
    )
    SYSTEM_METRICS_ENABLED: bool = Field(
        default=True,
        description="Instrument process/system metrics when METRICS_ENABLED is true",
    )
    ENVIRONMENT: str | None = Field(
        default=None,
        description="Deployment environment resource attribute (defaults from BaseConfig.ENVIRONMENT)",
    )
    FASTAPI_EXCLUDED_URLS: str | None = Field(
        default=None,
        description="Comma-separated URL patterns skipped by FastAPI instrumentation (e.g. health,docs)",
    )
    LOGS_LEVEL: str = Field(
        default="INFO",
        description=(
            "Minimum level for the LoggingHandler on the root logger. "
            "Records from opentelemetry.* loggers are filtered to avoid feedback loops. "
            "Prefer WARNING+ in production when LOGS_EXPORTER=otlp to limit export volume."
        ),
    )

    @model_validator(mode="after")
    def validate_otel_config(self) -> Self:
        """Validate signal switches, metrics exporters, and log level."""
        if self.IS_ENABLED and not (self.TRACES_ENABLED or self.METRICS_ENABLED or self.LOGS_ENABLED):
            raise ConfigurationError(
                operation="otel",
                reason="no_signals_enabled",
                additional_data={
                    "hint": "Enable at least one of TRACES_ENABLED, METRICS_ENABLED, or LOGS_ENABLED",
                },
            )

        if (
            self.METRICS_ENABLED
            and self.METRICS_EXPORTER == OtelMetricsExporter.PUSHGATEWAY
            and not (self.METRICS_PUSHGATEWAY_URL and self.METRICS_PUSHGATEWAY_URL.strip())
        ):
            raise ConfigurationError(
                operation="otel",
                reason="pushgateway_url_required",
                additional_data={
                    "hint": "Set METRICS_PUSHGATEWAY_URL when METRICS_EXPORTER=pushgateway",
                },
            )

        valid_levels = {"CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", "NOTSET"}
        if self.LOGS_LEVEL.upper() not in valid_levels:
            raise ConfigurationError(
                operation="otel",
                reason="invalid_logs_level",
                additional_data={"logs_level": self.LOGS_LEVEL, "allowed": sorted(valid_levels)},
            )
        return self

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

IS_ENABLED: bool = Field(
    default=False,
    description="Master switch for OpenTelemetry",
)

archipy.configs.config_template.OpentelemetryConfig.TRACES_ENABLED class-attribute instance-attribute 

TRACES_ENABLED: bool = Field(
    default=True,
    description="Export traces when OTel is enabled",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_ENABLED class-attribute instance-attribute 

METRICS_ENABLED: bool = Field(
    default=True,
    description="Export metrics when OTel is enabled",
)

archipy.configs.config_template.OpentelemetryConfig.LOGS_ENABLED class-attribute instance-attribute 

LOGS_ENABLED: bool = Field(
    default=True,
    description="Export logs when OTel is enabled",
)

archipy.configs.config_template.OpentelemetryConfig.LOGS_EXPORTER class-attribute instance-attribute 

LOGS_EXPORTER: OtelLogsExporter = Field(
    default=OtelLogsExporter.CONSOLE,
    description="Unique logs exporter when LOGS_ENABLED is true: console (stdout for INFO/DEBUG, stderr for WARNING+) or otlp",
)

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

SERVICE_NAME: str | None = Field(
    default=None, description="OTel resource service.name"
)

archipy.configs.config_template.OpentelemetryConfig.OTLP_ENDPOINT class-attribute instance-attribute 

OTLP_ENDPOINT: HttpUrl = Field(
    default=HttpUrl("http://localhost:4317"),
    description="Default OTLP collector endpoint (gRPC port 4317; HTTP/protobuf typically 4318). For http/protobuf, signal paths (/v1/traces, /v1/metrics, /v1/logs) are appended when the URL has no path. Override per signal with TRACES/METRICS/LOGS_ENDPOINT.",
)

archipy.configs.config_template.OpentelemetryConfig.TRACES_ENDPOINT class-attribute instance-attribute 

TRACES_ENDPOINT: HttpUrl | None = Field(
    default=None,
    description="Optional per-signal OTLP traces endpoint (overrides OTLP_ENDPOINT)",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_ENDPOINT class-attribute instance-attribute 

METRICS_ENDPOINT: HttpUrl | None = Field(
    default=None,
    description="Optional per-signal OTLP metrics endpoint (overrides OTLP_ENDPOINT)",
)

archipy.configs.config_template.OpentelemetryConfig.LOGS_ENDPOINT class-attribute instance-attribute 

LOGS_ENDPOINT: HttpUrl | None = Field(
    default=None,
    description="Optional per-signal OTLP logs endpoint (overrides OTLP_ENDPOINT)",
)

archipy.configs.config_template.OpentelemetryConfig.PROTOCOL class-attribute instance-attribute 

PROTOCOL: Literal["grpc", "http/protobuf"] = Field(
    default="grpc", description="OTLP transport protocol"
)

archipy.configs.config_template.OpentelemetryConfig.OTLP_HEADERS class-attribute instance-attribute 

OTLP_HEADERS: dict[str, str] = Field(
    default_factory=dict,
    description="Headers sent with each OTLP export (auth tokens, routing metadata)",
)

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

TIMEOUT: float = Field(
    default=10.0,
    ge=0.1,
    description="OTLP export timeout in seconds",
)

archipy.configs.config_template.OpentelemetryConfig.TRACES_SAMPLE_RATIO class-attribute instance-attribute 

TRACES_SAMPLE_RATIO: float = Field(
    default=0.1,
    ge=0.0,
    le=1.0,
    description="Parent-based trace ID ratio sampler (inbound sampled traces always continue)",
)

archipy.configs.config_template.OpentelemetryConfig.RESOURCE_ATTRIBUTES class-attribute instance-attribute 

RESOURCE_ATTRIBUTES: dict[str, str] = Field(
    default_factory=dict,
    description="Extra OTel resource attributes merged into the Resource",
)

archipy.configs.config_template.OpentelemetryConfig.METRIC_EXPORT_INTERVAL_MS class-attribute instance-attribute 

METRIC_EXPORT_INTERVAL_MS: int = Field(
    default=60000,
    ge=1000,
    description="Periodic OTLP metric export interval in milliseconds (used when METRICS_EXPORTER=otlp)",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_EXPORTER class-attribute instance-attribute 

METRICS_EXPORTER: OtelMetricsExporter = Field(
    default=OtelMetricsExporter.OTLP,
    description="Unique metrics exporter when METRICS_ENABLED is true (otlp, pull, or pushgateway)",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_PULL_HOST class-attribute instance-attribute 

METRICS_PULL_HOST: str = Field(
    default="0.0.0.0",
    description="Bind host for the metrics pull scrape server when METRICS_EXPORTER=pull",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_PULL_PORT class-attribute instance-attribute 

METRICS_PULL_PORT: int = Field(
    default=8200,
    ge=1,
    le=65535,
    description="Bind port for the metrics pull scrape server (/metrics)",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_PUSHGATEWAY_URL class-attribute instance-attribute 

METRICS_PUSHGATEWAY_URL: str | None = Field(
    default=None,
    description="Prometheus Pushgateway base URL when METRICS_EXPORTER=pushgateway (e.g. http://host:9091)",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_PUSHGATEWAY_JOB class-attribute instance-attribute 

METRICS_PUSHGATEWAY_JOB: str | None = Field(
    default=None,
    description="Pushgateway job label (defaults to SERVICE_NAME when unset)",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_PUSHGATEWAY_INTERVAL_SECONDS class-attribute instance-attribute 

METRICS_PUSHGATEWAY_INTERVAL_SECONDS: int = Field(
    default=60,
    ge=1,
    description="Interval between Pushgateway pushes when METRICS_EXPORTER=pushgateway",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_PUSHGATEWAY_TIMEOUT_SECONDS class-attribute instance-attribute 

METRICS_PUSHGATEWAY_TIMEOUT_SECONDS: float = Field(
    default=10.0,
    gt=0,
    description="HTTP timeout in seconds for Pushgateway push/delete calls",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_PUSHGATEWAY_GROUPING_KEY class-attribute instance-attribute 

METRICS_PUSHGATEWAY_GROUPING_KEY: dict[str, str] = Field(
    default_factory=dict,
    max_length=16,
    description="Extra Pushgateway grouping labels (instance, component, ...)",
)

archipy.configs.config_template.OpentelemetryConfig.METRICS_PUSHGATEWAY_DELETE_ON_SHUTDOWN class-attribute instance-attribute 

METRICS_PUSHGATEWAY_DELETE_ON_SHUTDOWN: bool = Field(
    default=True,
    description="Delete the Pushgateway grouping key on graceful OTel shutdown",
)

archipy.configs.config_template.OpentelemetryConfig.SYSTEM_METRICS_ENABLED class-attribute instance-attribute 

SYSTEM_METRICS_ENABLED: bool = Field(
    default=True,
    description="Instrument process/system metrics when METRICS_ENABLED is true",
)

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

ENVIRONMENT: str | None = Field(
    default=None,
    description="Deployment environment resource attribute (defaults from BaseConfig.ENVIRONMENT)",
)

archipy.configs.config_template.OpentelemetryConfig.FASTAPI_EXCLUDED_URLS class-attribute instance-attribute 

FASTAPI_EXCLUDED_URLS: str | None = Field(
    default=None,
    description="Comma-separated URL patterns skipped by FastAPI instrumentation (e.g. health,docs)",
)

archipy.configs.config_template.OpentelemetryConfig.LOGS_LEVEL class-attribute instance-attribute 

LOGS_LEVEL: str = Field(
    default="INFO",
    description="Minimum level for the LoggingHandler on the root logger. Records from opentelemetry.* loggers are filtered to avoid feedback loops. Prefer WARNING+ in production when LOGS_EXPORTER=otlp to limit export volume.",
)

archipy.configs.config_template.OpentelemetryConfig.validate_otel_config 

validate_otel_config() -> Self

Validate signal switches, metrics exporters, and log level.

Source code in archipy/configs/config_template.py 
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
@model_validator(mode="after")
def validate_otel_config(self) -> Self:
    """Validate signal switches, metrics exporters, and log level."""
    if self.IS_ENABLED and not (self.TRACES_ENABLED or self.METRICS_ENABLED or self.LOGS_ENABLED):
        raise ConfigurationError(
            operation="otel",
            reason="no_signals_enabled",
            additional_data={
                "hint": "Enable at least one of TRACES_ENABLED, METRICS_ENABLED, or LOGS_ENABLED",
            },
        )

    if (
        self.METRICS_ENABLED
        and self.METRICS_EXPORTER == OtelMetricsExporter.PUSHGATEWAY
        and not (self.METRICS_PUSHGATEWAY_URL and self.METRICS_PUSHGATEWAY_URL.strip())
    ):
        raise ConfigurationError(
            operation="otel",
            reason="pushgateway_url_required",
            additional_data={
                "hint": "Set METRICS_PUSHGATEWAY_URL when METRICS_EXPORTER=pushgateway",
            },
        )

    valid_levels = {"CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", "NOTSET"}
    if self.LOGS_LEVEL.upper() not in valid_levels:
        raise ConfigurationError(
            operation="otel",
            reason="invalid_logs_level",
            additional_data={"logs_level": self.LOGS_LEVEL, "allowed": sorted(valid_levels)},
        )
    return self

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 
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
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 
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
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 
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
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 
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
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 
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
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 (OTLP via global OTEL config)
    ENABLE_METRICS: bool = Field(
        default=False,
        description=(
            "Enable OTLP metrics export for Temporal workflows and activities "
            "(requires BaseConfig.OTEL.IS_ENABLED and METRICS_ENABLED; uses OTEL endpoint)"
        ),
    )

    # Client Connection Configuration
    CLIENT_IDENTITY: str | None = Field(
        default=None,
        description="Client identity string for Temporal server visibility",
    )
    API_KEY: str | None = Field(
        default=None,
        description="API key for Temporal Cloud authentication (ignored by local dev server)",
    )
    LAZY_CONNECT: bool = Field(
        default=False,
        description="Defer establishing the Temporal client connection until first use",
    )
    KEEP_ALIVE_INTERVAL_MS: int = Field(
        default=30000,
        ge=1,
        description="gRPC keep-alive ping interval in milliseconds",
    )
    KEEP_ALIVE_TIMEOUT_MS: int = Field(
        default=15000,
        ge=1,
        description="gRPC keep-alive timeout in milliseconds",
    )
    RPC_METADATA: dict[str, str] | None = Field(
        default=None,
        description="Custom RPC metadata headers sent with every client request",
    )
    CLIENT_RPC_RETRY_MAX_RETRIES: int = Field(
        default=10,
        ge=0,
        description="Maximum client-side RPC retry attempts",
    )
    CLIENT_RPC_RETRY_INITIAL_INTERVAL_MS: int = Field(
        default=100,
        ge=1,
        description="Initial backoff interval for client-side RPC retries in milliseconds",
    )
    CLIENT_RPC_RETRY_MAX_INTERVAL_MS: int = Field(
        default=5000,
        ge=1,
        description="Maximum backoff interval for client-side RPC retries in milliseconds",
    )

    # Worker Configuration
    WORKER_GRACEFUL_SHUTDOWN_SECONDS: int = Field(
        default=30,
        ge=0,
        description="Maximum time in seconds to wait for graceful worker shutdown",
    )
    WORKER_MAX_CACHED_WORKFLOWS: int = Field(
        default=1000,
        ge=0,
        description="Maximum number of workflow instances cached in worker memory",
    )
    WORKER_DEBUG_MODE: bool = Field(
        default=False,
        description="Enable worker debug mode (e.g. breakpoint support in workflow sandbox)",
    )
    WORKER_MAX_CONCURRENT_LOCAL_ACTIVITIES: int | None = Field(
        default=None,
        ge=1,
        description="Maximum concurrent local activities (None uses server default)",
    )
    WORKER_DISABLE_EAGER_ACTIVITY_EXECUTION: bool = Field(
        default=False,
        description="Disable eager activity execution on the worker",
    )

    # 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",
    )
    RETRY_INITIAL_INTERVAL: int = Field(
        default=1,
        ge=1,
        description="Initial retry interval in seconds for workflow and activity retry policies",
    )
    RETRY_NON_RETRYABLE_ERROR_TYPES: list[str] | None = Field(
        default=None,
        description="Error type names that should not be retried by workflow/activity retry policies",
    )

    @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) != TLS_CERT_PARTS_REQUIRED:
            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

    @model_validator(mode="after")
    def validate_keep_alive_configuration(self) -> Self:
        """Validate keep-alive timeout is less than the keep-alive interval."""
        if self.KEEP_ALIVE_TIMEOUT_MS >= self.KEEP_ALIVE_INTERVAL_MS:
            raise InvalidArgumentError()

        return self

    @model_validator(mode="after")
    def validate_rpc_metadata(self) -> Self:
        """Validate RPC metadata keys and values are non-empty strings."""
        if self.RPC_METADATA is None:
            return self

        for key, value in self.RPC_METADATA.items():
            if not key or not value:
                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 OTLP metrics export for Temporal workflows and activities (requires BaseConfig.OTEL.IS_ENABLED and METRICS_ENABLED; uses OTEL endpoint)",
)

archipy.configs.config_template.TemporalConfig.CLIENT_IDENTITY class-attribute instance-attribute 

CLIENT_IDENTITY: str | None = Field(
    default=None,
    description="Client identity string for Temporal server visibility",
)

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

API_KEY: str | None = Field(
    default=None,
    description="API key for Temporal Cloud authentication (ignored by local dev server)",
)

archipy.configs.config_template.TemporalConfig.LAZY_CONNECT class-attribute instance-attribute 

LAZY_CONNECT: bool = Field(
    default=False,
    description="Defer establishing the Temporal client connection until first use",
)

archipy.configs.config_template.TemporalConfig.KEEP_ALIVE_INTERVAL_MS class-attribute instance-attribute 

KEEP_ALIVE_INTERVAL_MS: int = Field(
    default=30000,
    ge=1,
    description="gRPC keep-alive ping interval in milliseconds",
)

archipy.configs.config_template.TemporalConfig.KEEP_ALIVE_TIMEOUT_MS class-attribute instance-attribute 

KEEP_ALIVE_TIMEOUT_MS: int = Field(
    default=15000,
    ge=1,
    description="gRPC keep-alive timeout in milliseconds",
)

archipy.configs.config_template.TemporalConfig.RPC_METADATA class-attribute instance-attribute 

RPC_METADATA: dict[str, str] | None = Field(
    default=None,
    description="Custom RPC metadata headers sent with every client request",
)

archipy.configs.config_template.TemporalConfig.CLIENT_RPC_RETRY_MAX_RETRIES class-attribute instance-attribute 

CLIENT_RPC_RETRY_MAX_RETRIES: int = Field(
    default=10,
    ge=0,
    description="Maximum client-side RPC retry attempts",
)

archipy.configs.config_template.TemporalConfig.CLIENT_RPC_RETRY_INITIAL_INTERVAL_MS class-attribute instance-attribute 

CLIENT_RPC_RETRY_INITIAL_INTERVAL_MS: int = Field(
    default=100,
    ge=1,
    description="Initial backoff interval for client-side RPC retries in milliseconds",
)

archipy.configs.config_template.TemporalConfig.CLIENT_RPC_RETRY_MAX_INTERVAL_MS class-attribute instance-attribute 

CLIENT_RPC_RETRY_MAX_INTERVAL_MS: int = Field(
    default=5000,
    ge=1,
    description="Maximum backoff interval for client-side RPC retries in milliseconds",
)

archipy.configs.config_template.TemporalConfig.WORKER_GRACEFUL_SHUTDOWN_SECONDS class-attribute instance-attribute 

WORKER_GRACEFUL_SHUTDOWN_SECONDS: int = Field(
    default=30,
    ge=0,
    description="Maximum time in seconds to wait for graceful worker shutdown",
)

archipy.configs.config_template.TemporalConfig.WORKER_MAX_CACHED_WORKFLOWS class-attribute instance-attribute 

WORKER_MAX_CACHED_WORKFLOWS: int = Field(
    default=1000,
    ge=0,
    description="Maximum number of workflow instances cached in worker memory",
)

archipy.configs.config_template.TemporalConfig.WORKER_DEBUG_MODE class-attribute instance-attribute 

WORKER_DEBUG_MODE: bool = Field(
    default=False,
    description="Enable worker debug mode (e.g. breakpoint support in workflow sandbox)",
)

archipy.configs.config_template.TemporalConfig.WORKER_MAX_CONCURRENT_LOCAL_ACTIVITIES class-attribute instance-attribute 

WORKER_MAX_CONCURRENT_LOCAL_ACTIVITIES: int | None = Field(
    default=None,
    ge=1,
    description="Maximum concurrent local activities (None uses server default)",
)

archipy.configs.config_template.TemporalConfig.WORKER_DISABLE_EAGER_ACTIVITY_EXECUTION class-attribute instance-attribute 

WORKER_DISABLE_EAGER_ACTIVITY_EXECUTION: bool = Field(
    default=False,
    description="Disable eager activity execution on the worker",
)

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.RETRY_INITIAL_INTERVAL class-attribute instance-attribute 

RETRY_INITIAL_INTERVAL: int = Field(
    default=1,
    ge=1,
    description="Initial retry interval in seconds for workflow and activity retry policies",
)

archipy.configs.config_template.TemporalConfig.RETRY_NON_RETRYABLE_ERROR_TYPES class-attribute instance-attribute 

RETRY_NON_RETRYABLE_ERROR_TYPES: list[str] | None = Field(
    default=None,
    description="Error type names that should not be retried by workflow/activity retry policies",
)

archipy.configs.config_template.TemporalConfig.validate_tls_configuration 

validate_tls_configuration() -> Self

Validate TLS configuration consistency.

Source code in archipy/configs/config_template.py 
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
@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) != TLS_CERT_PARTS_REQUIRED:
        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 
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
@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.validate_keep_alive_configuration 

validate_keep_alive_configuration() -> Self

Validate keep-alive timeout is less than the keep-alive interval.

Source code in archipy/configs/config_template.py 
1432
1433
1434
1435
1436
1437
1438
@model_validator(mode="after")
def validate_keep_alive_configuration(self) -> Self:
    """Validate keep-alive timeout is less than the keep-alive interval."""
    if self.KEEP_ALIVE_TIMEOUT_MS >= self.KEEP_ALIVE_INTERVAL_MS:
        raise InvalidArgumentError()

    return self

archipy.configs.config_template.TemporalConfig.validate_rpc_metadata 

validate_rpc_metadata() -> Self

Validate RPC metadata keys and values are non-empty strings.

Source code in archipy/configs/config_template.py 
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
@model_validator(mode="after")
def validate_rpc_metadata(self) -> Self:
    """Validate RPC metadata keys and values are non-empty strings."""
    if self.RPC_METADATA is None:
        return self

    for key, value in self.RPC_METADATA.items():
        if not key or not value:
            raise InvalidArgumentError()

    return self

options: show_root_toc_entry: false heading_level: 3

Environment Type

Environment type enumeration for ArchiPy configs.

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 
 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
95
96
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 
46
47
48
49
50
51
52
53
@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 
55
56
57
58
59
60
61
62
@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 
64
65
66
67
68
69
70
71
@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 
73
74
75
76
77
78
79
80
81
82
@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 
84
85
86
87
88
89
90
91
92
93
94
95
96
@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