Configs

Overview

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

Quick Start

from archipy.configs.base_config import BaseConfig

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

API Stability

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

Core Classes

BaseConfig

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

Key Features:

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

Classes:

Name	Description
BaseConfig	Base configuration class for ArchiPy applications.

BaseConfig

Base configuration class for ArchiPy applications.

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

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

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

Attributes:

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

Examples:

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

Methods:

Name	Description
settings_customise_sources	Customize the settings sources priority order.
customize	Customize configuration after loading.
global_config	Retrieves the global configuration instance.
set_global	Sets the global configuration instance.

Source code in archipy/configs/base_config.py

class BaseConfig[R](BaseSettings):
    """Base configuration class for ArchiPy applications.

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

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

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

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

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

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

    __global_config: BaseConfig | None = None

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

        This method defines the priority order for configuration sources.

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

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

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

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

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

        if self.SENTRY.ENVIRONMENT is None:
            self.SENTRY.ENVIRONMENT = self.ENVIRONMENT

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

        Returns:
            BaseConfig: The global configuration instance.

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

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

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

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

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

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

settings_customise_sources classmethod

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

Customize the settings sources priority order.

This method defines the priority order for configuration sources.

Parameters:

Name	Type	Description	Default
settings_cls	type[BaseSettings]	The settings class	required
init_settings	PydanticBaseSettingsSource	Settings from initialization values	required
env_settings	PydanticBaseSettingsSource	Settings from environment variables	required
dotenv_settings	PydanticBaseSettingsSource	Settings from .env file	required
file_secret_settings	PydanticBaseSettingsSource	Settings from secret files	required

Returns:

Type	Description
tuple[PydanticBaseSettingsSource, ...]	A tuple of configuration sources in priority order

Source code in archipy/configs/base_config.py

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

    This method defines the priority order for configuration sources.

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

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

customize

customize() -> None

Customize configuration after loading.

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

Source code in archipy/configs/base_config.py

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

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

    if self.SENTRY.ENVIRONMENT is None:
        self.SENTRY.ENVIRONMENT = self.ENVIRONMENT

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

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

    Returns:
        BaseConfig: The global configuration instance.

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

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

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

@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

Configuration settings for SQLAlchemy ORM.

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

Source code in archipy/configs/config_template.py

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

options: show_root_toc_entry: false heading_level: 3

Configuration settings for PostgreSQL SQLAlchemy ORM.

Extends SQLAlchemyConfig with PostgreSQL-specific settings and URL building.

Methods:

Name	Description
build_connection_url	Build and populate DB_URL if not provided but all component parts are present.
extract_connection_parts	Extract connection parts from DB_URL if provided but component parts are missing.

Source code in archipy/configs/config_template.py

class PostgresSQLAlchemyConfig(SQLAlchemyConfig):
    """Configuration settings for PostgreSQL SQLAlchemy ORM.

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

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

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

        Returns:
            Self: The updated configuration instance.

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

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

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

        Returns:
            Self: The updated configuration instance.

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

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

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

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

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

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

        return self

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

@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

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

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

    Returns:
        Self: The updated configuration instance.

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

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

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

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

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

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

    return self

options: show_root_toc_entry: false heading_level: 3

Configuration settings for SQLite SQLAlchemy ORM.

Extends SQLAlchemyConfig with SQLite-specific settings.

Source code in archipy/configs/config_template.py

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

options: show_root_toc_entry: false heading_level: 3

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.

Methods:

Name	Description
validate_isolation_level	Validate that isolation level is READ COMMITTED for StarRocks.

Source code in archipy/configs/config_template.py

class StarRocksSQLAlchemyConfig(SQLAlchemyConfig):
    """Configuration settings for Starrocks SQLAlchemy ORM.

    Extends SQLAlchemyConfig with Starrocks-specific settings.

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

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

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

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

        Args:
            v: The isolation level value to validate.

        Returns:
            The validated isolation level.

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

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

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

    Args:
        v: The isolation level value to validate.

    Returns:
        The validated isolation level.

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

options: show_root_toc_entry: false heading_level: 3

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.

Methods:

Name	Description
validate_contact_points	Validate that at least one contact point is provided.
validate_authentication	Validate that both username and password are provided together.
validate_connection_pool	Validate connection pool configuration.
validate_replication_config	Validate replication configuration.

Source code in archipy/configs/config_template.py

class ScyllaDBConfig(BaseModel):
    """Configuration settings for ScyllaDB/Cassandra connections and operations.

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

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

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

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

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

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

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

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

validate_contact_points

validate_contact_points() -> Self

Validate that at least one contact point is provided.

Source code in archipy/configs/config_template.py

@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

validate_authentication

validate_authentication() -> Self

Validate that both username and password are provided together.

Source code in archipy/configs/config_template.py

@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

validate_connection_pool

validate_connection_pool() -> Self

Validate connection pool configuration.

Source code in archipy/configs/config_template.py

@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

validate_replication_config

validate_replication_config() -> Self

Validate replication configuration.

Source code in archipy/configs/config_template.py

@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

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.

Methods:

Name	Description
validate_tls_settings	Validate TLS-related settings to ensure compatibility.
validate_sniffing_settings	Warn if sniffing is enabled with a load balancer.

Source code in archipy/configs/config_template.py

class ElasticsearchConfig(BaseModel):
    """Configuration settings for Elasticsearch connections and operations.

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

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

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

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

validate_tls_settings

validate_tls_settings() -> Self

Validate TLS-related settings to ensure compatibility.

Source code in archipy/configs/config_template.py

@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

validate_sniffing_settings

validate_sniffing_settings() -> Self

Warn if sniffing is enabled with a load balancer.

Source code in archipy/configs/config_template.py

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

options: show_root_toc_entry: false heading_level: 3

Configuration settings for Elasticsearch APM (Application Performance Monitoring).

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

Source code in archipy/configs/config_template.py

class ElasticsearchAPMConfig(BaseModel):
    """Configuration settings for Elasticsearch APM (Application Performance Monitoring).

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

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

options: show_root_toc_entry: false heading_level: 3

Service Configs

Configuration settings for Redis cache integration.

Supports standalone, sentinel, and cluster deployments.

Methods:

Name	Description
validate_mode_configuration	Validate mode-specific configuration.

Source code in archipy/configs/config_template.py

class RedisConfig(BaseModel):
    """Configuration settings for Redis cache integration.

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

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

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

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

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

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

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

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

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

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

        return self

validate_mode_configuration

validate_mode_configuration() -> Self

Validate mode-specific configuration.

Source code in archipy/configs/config_template.py

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

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

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

    return self

options: show_root_toc_entry: false heading_level: 3

Configuration settings for Apache Kafka integration.

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

Methods:

Name	Description
validate_security_settings	Validate security-related settings for Kafka configuration.
validate_consumer_settings	Validate consumer-specific settings for Kafka configuration.
validate_idempotence_and_transactions	Validate idempotence and transaction settings for Kafka configuration.

Source code in archipy/configs/config_template.py

class KafkaConfig(BaseModel):
    """Configuration settings for Apache Kafka integration.

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

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

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

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

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

        Returns:
            KafkaConfig: The validated configuration instance.

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

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

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

        Returns:
            KafkaConfig: The validated configuration instance.

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

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

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

        Returns:
            KafkaConfig: The validated configuration instance.

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

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

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

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

    Returns:
        KafkaConfig: The validated configuration instance.

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

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

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

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

    Returns:
        KafkaConfig: The validated configuration instance.

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

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

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

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

    Returns:
        KafkaConfig: The validated configuration instance.

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

options: show_root_toc_entry: false heading_level: 3

Configuration settings for email service integration.

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

Source code in archipy/configs/config_template.py

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

options: show_root_toc_entry: false heading_level: 3

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

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

options: show_root_toc_entry: false heading_level: 3

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

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"

options: show_root_toc_entry: false heading_level: 3

Web Framework Configs

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

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

options: show_root_toc_entry: false heading_level: 3

Configuration settings for gRPC services.

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

Methods:

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

Source code in archipy/configs/config_template.py

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

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

@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

Configuration settings for Sentry error tracking integration.

Controls Sentry client behavior, including DSN, sampling rates, PII, breadcrumbs, stack traces, profiling, and integration defaults.

Source code in archipy/configs/config_template.py

class SentryConfig(BaseModel):
    """Configuration settings for Sentry error tracking integration.

    Controls Sentry client behavior, including DSN, sampling rates, PII, breadcrumbs,
    stack traces, profiling, and integration defaults.
    """

    IS_ENABLED: bool = Field(default=False, description="Whether Sentry is enabled")
    DSN: str | None = Field(default=None, description="Sentry DSN for error reporting")
    DEBUG: bool = Field(default=False, description="Whether to enable debug mode")
    RELEASE: str = Field(default="", description="Application release version")
    SAMPLE_RATE: float = Field(default=1.0, description="Error sampling rate (0.0 to 1.0)")
    TRACES_SAMPLE_RATE: float = Field(default=0.0, description="Performance monitoring sampling rate (0.0 to 1.0)")
    SEND_DEFAULT_PII: bool = Field(
        default=False,
        description="Whether to send personally identifiable information (e.g. IP, user) to Sentry",
    )
    MAX_BREADCRUMBS: int = Field(default=100, description="Maximum number of breadcrumbs per event")
    ATTACH_STACKTRACE: bool = Field(
        default=False,
        description="Whether to attach a stack trace to all messages (not only exceptions)",
    )
    SERVER_NAME: str | None = Field(default=None, description="Override the reported server hostname")
    IN_APP_INCLUDE: list[str] = Field(
        default_factory=list,
        description="Module path prefixes to mark as in-app in stack traces",
    )
    IN_APP_EXCLUDE: list[str] = Field(
        default_factory=list,
        description="Module path prefixes to exclude from in-app marking",
    )
    PROFILES_SAMPLE_RATE: float = Field(
        default=0.0,
        description="Continuous profiling sample rate (0.0 to 1.0)",
    )
    IGNORE_ERRORS: list[str] = Field(
        default_factory=list,
        description="Exception type names (fully qualified or short) to ignore",
    )
    SHUTDOWN_TIMEOUT: int = Field(default=2, description="Seconds to wait for pending events on client shutdown")
    DEFAULT_INTEGRATIONS: bool = Field(
        default=True,
        description="Whether to enable Sentry default integrations automatically",
    )
    ENVIRONMENT: str | None = Field(default=None, description="Sentry environment name")

options: show_root_toc_entry: false heading_level: 3

Configuration settings for Prometheus metrics integration.

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

Source code in archipy/configs/config_template.py

class PrometheusConfig(BaseModel):
    """Configuration settings for Prometheus metrics integration.

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

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

options: show_root_toc_entry: false heading_level: 3

Payment Configs

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

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

options: show_root_toc_entry: false heading_level: 3

Application Configs

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

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

options: show_root_toc_entry: false heading_level: 3

Configuration settings for file handling capabilities.

Controls file link security, expiration policies, and file type restrictions.

Source code in archipy/configs/config_template.py

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

options: show_root_toc_entry: false heading_level: 3

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

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

options: show_root_toc_entry: false heading_level: 3

Configuration settings for Temporal workflow engine integration.

Controls connection parameters, security settings, and timeout configurations for Temporal workflow orchestration services.

Methods:

Name	Description
validate_tls_configuration	Validate TLS configuration consistency.
validate_timeout_hierarchy	Validate timeout configuration hierarchy.

Source code in archipy/configs/config_template.py

class TemporalConfig(BaseModel):
    """Configuration settings for Temporal workflow engine integration.

    Controls connection parameters, security settings, and timeout configurations
    for Temporal workflow orchestration services.
    """

    HOST: str = Field(default="localhost", description="Temporal server host address")
    PORT: int = Field(default=7233, ge=1, le=65535, description="Temporal server port number")
    NAMESPACE: str = Field(default="default", description="Temporal namespace for workflow isolation")
    TASK_QUEUE: str = Field(default="task-queue", description="Default task queue name")

    # Metrics Configuration
    ENABLE_METRICS: bool = Field(
        default=False,
        description="Enable Prometheus metrics collection for Temporal workflows and activities",
    )
    METRICS_PORT: int = Field(
        default=8201,
        ge=1,
        le=65535,
        description="Port for Temporal Prometheus metrics endpoint (separate from main Prometheus port)",
    )

    # TLS Configuration
    TLS_CA_CERT: str | None = Field(default=None, description="Path to TLS CA certificate")
    TLS_CLIENT_CERT: str | None = Field(default=None, description="Path to TLS client certificate")
    TLS_CLIENT_KEY: str | None = Field(default=None, description="Path to TLS client private key")

    # Workflow Timeout Settings
    WORKFLOW_EXECUTION_TIMEOUT: int = Field(
        default=300,
        ge=1,
        description="Maximum workflow execution time in seconds",
    )
    WORKFLOW_RUN_TIMEOUT: int = Field(
        default=60,
        ge=1,
        description="Maximum single workflow run time in seconds",
    )
    WORKFLOW_TASK_TIMEOUT: int = Field(
        default=30,
        ge=1,
        description="Maximum workflow task processing time in seconds",
    )

    # Activity Timeout Settings
    ACTIVITY_START_TO_CLOSE_TIMEOUT: int = Field(
        default=30,
        ge=1,
        description="Maximum activity execution time in seconds",
    )
    ACTIVITY_HEARTBEAT_TIMEOUT: int = Field(
        default=10,
        ge=1,
        description="Activity heartbeat timeout in seconds",
    )

    # Retry Configuration
    RETRY_MAXIMUM_ATTEMPTS: int = Field(
        default=3,
        ge=1,
        description="Maximum retry attempts for failed activities",
    )
    RETRY_BACKOFF_COEFFICIENT: float = Field(
        default=2.0,
        ge=1.0,
        description="Backoff multiplier for retry delays",
    )
    RETRY_MAXIMUM_INTERVAL: int = Field(
        default=60,
        ge=1,
        description="Maximum retry interval in seconds",
    )

    @model_validator(mode="after")
    def validate_tls_configuration(self) -> Self:
        """Validate TLS configuration consistency."""
        tls_fields = [self.TLS_CA_CERT, self.TLS_CLIENT_CERT, self.TLS_CLIENT_KEY]
        tls_provided = [field for field in tls_fields if field is not None]

        if len(tls_provided) > 0 and len(tls_provided) != 3:
            raise InvalidArgumentError()

        return self

    @model_validator(mode="after")
    def validate_timeout_hierarchy(self) -> Self:
        """Validate timeout configuration hierarchy."""
        if self.WORKFLOW_RUN_TIMEOUT >= self.WORKFLOW_EXECUTION_TIMEOUT:
            raise InvalidArgumentError()

        if self.WORKFLOW_TASK_TIMEOUT >= self.WORKFLOW_RUN_TIMEOUT:
            raise InvalidArgumentError()

        return self

validate_tls_configuration

validate_tls_configuration() -> Self

Validate TLS configuration consistency.

Source code in archipy/configs/config_template.py

@model_validator(mode="after")
def validate_tls_configuration(self) -> Self:
    """Validate TLS configuration consistency."""
    tls_fields = [self.TLS_CA_CERT, self.TLS_CLIENT_CERT, self.TLS_CLIENT_KEY]
    tls_provided = [field for field in tls_fields if field is not None]

    if len(tls_provided) > 0 and len(tls_provided) != 3:
        raise InvalidArgumentError()

    return self

validate_timeout_hierarchy

validate_timeout_hierarchy() -> Self

Validate timeout configuration hierarchy.

Source code in archipy/configs/config_template.py

@model_validator(mode="after")
def validate_timeout_hierarchy(self) -> Self:
    """Validate timeout configuration hierarchy."""
    if self.WORKFLOW_RUN_TIMEOUT >= self.WORKFLOW_EXECUTION_TIMEOUT:
        raise InvalidArgumentError()

    if self.WORKFLOW_TASK_TIMEOUT >= self.WORKFLOW_RUN_TIMEOUT:
        raise InvalidArgumentError()

    return self

options: show_root_toc_entry: false heading_level: 3

Environment Type

Classes:

Name	Description
EnvironmentType	Enum representing different application environments.

EnvironmentType

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

Methods:

Name	Description
is_local	Check if the environment is LOCAL.
is_dev	Check if the environment is DEV.
is_test	Check if the environment is a testing environment.
is_production	Check if the environment is a production environment.
log_level	Get the appropriate logging level for this environment.

Source code in archipy/configs/environment_type.py

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

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

@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

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

@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

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

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

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

@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

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

@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