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

Base configuration system for ArchiPy applications.

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

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

archipy.configs.base_config.R module-attribute

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

archipy.configs.base_config.BaseConfig

Bases: BaseSettings

Base configuration class for ArchiPy applications.

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

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

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

Attributes:

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

Examples:

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

Source code in archipy/configs/base_config.py

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
        FASTAPI_RATE_LIMIT (FastAPIRateLimitConfig): FastAPI REST rate limiting settings
        FILE (FileConfig): File handling configuration
        GRPC (GrpcConfig): gRPC service configuration
        GRPC_RATE_LIMIT (GrpcRateLimitConfig): gRPC server rate limiting settings
        KAFKA (KafkaConfig): Kafka integration configuration
        KEYCLOAK (KeycloakConfig): Keycloak integration configuration
        LANGUAGE (LanguageType): Application default language
        MINIO (MinioConfig): MinIO object storage configuration
        PARSIAN_SHAPARAK (ParsianShaparakConfig): Parsian Shaparak payment gateway configuration
        POSTGRES_SQLALCHEMY (PostgresSQLAlchemyConfig): PostgreSQL SQLAlchemy configuration
        PROMETHEUS (PrometheusConfig): Prometheus metrics configuration
        REDIS (RedisConfig): Redis cache configuration
        SAMAN_SHAPARAK (SamanShaparakConfig): Saman Shaparak (SEP) payment gateway 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()
    FASTAPI_RATE_LIMIT: FastAPIRateLimitConfig = FastAPIRateLimitConfig()
    FILE: FileConfig = FileConfig()
    GRPC: GrpcConfig = GrpcConfig()
    GRPC_RATE_LIMIT: GrpcRateLimitConfig = GrpcRateLimitConfig()
    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.
        It is called automatically by `set_global()`.

        Examples:
            >>> class MyAppConfig(BaseConfig):
            ...     def customize(self) -> None:
            ...         super().customize()
            ...         self.REDIS.MASTER_HOST = "redis.internal"
            >>> config = MyAppConfig()
            >>> BaseConfig.set_global(config)  # calls customize() automatically
        """
        if self.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

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

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

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

AUTH: AuthConfig = AuthConfig()

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

DATETIME: DatetimeConfig = DatetimeConfig()

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

ELASTIC: ElasticsearchConfig = ElasticsearchConfig()

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

ELASTIC_APM: ElasticsearchAPMConfig = (
    ElasticsearchAPMConfig()
)

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

EMAIL: EmailConfig = EmailConfig()

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

ENVIRONMENT: EnvironmentType = EnvironmentType.LOCAL

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

FASTAPI: FastAPIConfig = FastAPIConfig()

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

FASTAPI_RATE_LIMIT: FastAPIRateLimitConfig = (
    FastAPIRateLimitConfig()
)

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

FILE: FileConfig = FileConfig()

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

GRPC: GrpcConfig = GrpcConfig()

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

GRPC_RATE_LIMIT: GrpcRateLimitConfig = GrpcRateLimitConfig()

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

KAFKA: KafkaConfig = KafkaConfig()

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

KEYCLOAK: KeycloakConfig = KeycloakConfig()

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

MINIO: MinioConfig = MinioConfig()

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

PARSIAN_SHAPARAK: ParsianShaparakConfig = (
    ParsianShaparakConfig()
)

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

SAMAN_SHAPARAK: SamanShaparakConfig = SamanShaparakConfig()

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

PROMETHEUS: PrometheusConfig = PrometheusConfig()

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

REDIS: RedisConfig = RedisConfig()

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

SCYLLADB: ScyllaDBConfig = ScyllaDBConfig()

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

SENTRY: SentryConfig = SentryConfig()

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

SQLALCHEMY: SQLAlchemyConfig = SQLAlchemyConfig()

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

STARROCKS_SQLALCHEMY: StarRocksSQLAlchemyConfig = (
    StarRocksSQLAlchemyConfig()
)

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

POSTGRES_SQLALCHEMY: PostgresSQLAlchemyConfig = (
    PostgresSQLAlchemyConfig()
)

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

SQLITE_SQLALCHEMY: SQLiteSQLAlchemyConfig = (
    SQLiteSQLAlchemyConfig()
)

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

TEMPORAL: TemporalConfig = TemporalConfig()

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

LANGUAGE: LanguageType = LanguageType.FA

archipy.configs.base_config.BaseConfig.settings_customise_sources classmethod

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

Customize the settings sources priority order.

This method defines the priority order for configuration sources.

Parameters:

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

Returns:

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

Source code in archipy/configs/base_config.py

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

    This method defines the priority order for configuration sources.

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

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

archipy.configs.base_config.BaseConfig.customize

customize() -> None

Customize configuration after loading.

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

Examples:

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

Source code in archipy/configs/base_config.py

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

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

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

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

archipy.configs.base_config.BaseConfig.global_config classmethod

global_config() -> BaseConfig

Retrieves the global configuration instance.

Returns:

Name	Type	Description
BaseConfig	BaseConfig	The global configuration instance.

Raises:

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

Examples:

>>> config = BaseConfig.global_config()
>>> redis_host = config.REDIS.MASTER_HOST

Source code in archipy/configs/base_config.py

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

    Returns:
        BaseConfig: The global configuration instance.

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

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

archipy.configs.base_config.BaseConfig.set_global classmethod

set_global(config: BaseConfig) -> None

Sets the global configuration instance.

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

Parameters:

Name	Type	Description	Default
config	BaseConfig	The configuration instance to use globally.	required

Examples:

>>> my_config = MyAppConfig(BaseConfig)
>>> BaseConfig.set_global(my_config)

Source code in archipy/configs/base_config.py

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

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

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

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

options: show_root_toc_entry: false heading_level: 3 members_order: alphabetical

Config Templates

For practical examples, see the Configuration Management Guide.

Database Configs

Bases: BaseModel

Configuration settings for SQLAlchemy ORM.

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

Source code in archipy/configs/config_template.py

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

options: show_root_toc_entry: false heading_level: 3

Bases: SQLAlchemyConfig

Configuration settings for PostgreSQL SQLAlchemy ORM.

Extends SQLAlchemyConfig with PostgreSQL-specific settings and URL building.

Source code in archipy/configs/config_template.py

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

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

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

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

        Returns:
            Self: The updated configuration instance.

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

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

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

        Returns:
            Self: The updated configuration instance.

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

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

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

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

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

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

        return self

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

archipy.configs.config_template.PostgresSQLAlchemyConfig.build_connection_url

build_connection_url() -> Self

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

Returns:

Name	Type	Description
Self	Self	The updated configuration instance.

Raises:

Type	Description
ValueError	If required connection parameters are missing.

Source code in archipy/configs/config_template.py

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

    Returns:
        Self: The updated configuration instance.

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

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

archipy.configs.config_template.PostgresSQLAlchemyConfig.extract_connection_parts

extract_connection_parts() -> Self

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

Returns:

Name	Type	Description
Self	Self	The updated configuration instance.

Raises:

Type	Description
ValueError	If the connection URL is invalid.

Source code in archipy/configs/config_template.py

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

    Returns:
        Self: The updated configuration instance.

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

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

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

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

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

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

    return self

options: show_root_toc_entry: false heading_level: 3

Bases: SQLAlchemyConfig

Configuration settings for SQLite SQLAlchemy ORM.

Extends SQLAlchemyConfig with SQLite-specific settings.

Source code in archipy/configs/config_template.py

class SQLiteSQLAlchemyConfig(SQLAlchemyConfig):
    """Configuration settings for SQLite SQLAlchemy ORM.

    Extends SQLAlchemyConfig with SQLite-specific settings.
    """

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

options: show_root_toc_entry: false heading_level: 3

Bases: SQLAlchemyConfig

Configuration settings for Starrocks SQLAlchemy ORM.

Extends SQLAlchemyConfig with Starrocks-specific settings.

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

Source code in archipy/configs/config_template.py

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

    Extends SQLAlchemyConfig with Starrocks-specific settings.

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

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

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

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

        Args:
            v: The isolation level value to validate.

        Returns:
            The validated isolation level.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

archipy.configs.config_template.StarRocksSQLAlchemyConfig.validate_isolation_level classmethod

validate_isolation_level(v: str) -> str

Validate that isolation level is READ COMMITTED for StarRocks.

Parameters:

Name	Type	Description	Default
v	str	The isolation level value to validate.	required

Returns:

Type	Description
str	The validated isolation level.

Raises:

Type	Description
ValueError	If the isolation level is not READ COMMITTED.

Source code in archipy/configs/config_template.py

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

    Args:
        v: The isolation level value to validate.

    Returns:
        The validated isolation level.

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

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for ScyllaDB/Cassandra connections and operations.

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

Source code in archipy/configs/config_template.py

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

archipy.configs.config_template.ScyllaDBConfig.validate_contact_points

validate_contact_points() -> Self

Validate that at least one contact point is provided.

Source code in archipy/configs/config_template.py

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

archipy.configs.config_template.ScyllaDBConfig.validate_authentication

validate_authentication() -> Self

Validate that both username and password are provided together.

Source code in archipy/configs/config_template.py

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

archipy.configs.config_template.ScyllaDBConfig.validate_connection_pool

validate_connection_pool() -> Self

Validate connection pool configuration.

Source code in archipy/configs/config_template.py

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

archipy.configs.config_template.ScyllaDBConfig.validate_replication_config

validate_replication_config() -> Self

Validate replication configuration.

Source code in archipy/configs/config_template.py

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

options: show_root_toc_entry: false heading_level: 3

Search & Analytics Configs

Bases: BaseModel

Configuration settings for Elasticsearch connections and operations.

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

Source code in archipy/configs/config_template.py

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

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

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

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

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

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

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

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

HTTP_USER_NAME: str | None = None

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

HTTP_PASSWORD: SecretStr | None = None

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

API_KEY: str | None = None

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

API_SECRET: SecretStr | None = None

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

archipy.configs.config_template.ElasticsearchConfig.validate_tls_settings

validate_tls_settings() -> Self

Validate TLS-related settings to ensure compatibility.

Source code in archipy/configs/config_template.py

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

archipy.configs.config_template.ElasticsearchConfig.validate_sniffing_settings

validate_sniffing_settings() -> Self

Warn if sniffing is enabled with a load balancer.

Source code in archipy/configs/config_template.py

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

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for Elasticsearch APM (Application Performance Monitoring).

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

Source code in archipy/configs/config_template.py

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

options: show_root_toc_entry: false heading_level: 3

Service Configs

Bases: BaseModel

Configuration settings for Redis cache integration.

Supports standalone, sentinel, and cluster deployments.

Source code in archipy/configs/config_template.py

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

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

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

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

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

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

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

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

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

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

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

        return self

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

archipy.configs.config_template.RedisConfig.validate_mode_configuration

validate_mode_configuration() -> Self

Validate mode-specific configuration.

Source code in archipy/configs/config_template.py

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

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

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

    return self

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for Apache Kafka integration.

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

Source code in archipy/configs/config_template.py

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

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

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

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

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

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

        Returns:
            KafkaConfig: The validated configuration instance.

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

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

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

        Returns:
            KafkaConfig: The validated configuration instance.

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

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

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

        Returns:
            KafkaConfig: The validated configuration instance.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

archipy.configs.config_template.KafkaConfig.validate_security_settings

validate_security_settings() -> KafkaConfig

Validate security-related settings for Kafka configuration.

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

Returns:

Name	Type	Description
KafkaConfig	KafkaConfig	The validated configuration instance.

Raises:

Type	Description
ValueError	If SASL authentication is incomplete.

Source code in archipy/configs/config_template.py

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

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

    Returns:
        KafkaConfig: The validated configuration instance.

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

archipy.configs.config_template.KafkaConfig.validate_consumer_settings

validate_consumer_settings() -> KafkaConfig

Validate consumer-specific settings for Kafka configuration.

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

Returns:

Name	Type	Description
KafkaConfig	KafkaConfig	The validated configuration instance.

Raises:

Type	Description
ValueError	If consumer settings are incompatible.

Source code in archipy/configs/config_template.py

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

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

    Returns:
        KafkaConfig: The validated configuration instance.

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

archipy.configs.config_template.KafkaConfig.validate_idempotence_and_transactions

validate_idempotence_and_transactions() -> KafkaConfig

Validate idempotence and transaction settings for Kafka configuration.

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

Returns:

Name	Type	Description
KafkaConfig	KafkaConfig	The validated configuration instance.

Raises:

Type	Description
ValueError	If idempotence or transaction settings are invalid.

Source code in archipy/configs/config_template.py

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

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

    Returns:
        KafkaConfig: The validated configuration instance.

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

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for email service integration.

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

Source code in archipy/configs/config_template.py

class EmailConfig(BaseModel):
    """Configuration settings for email service integration.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for MinIO/S3 object storage integration.

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

Source code in archipy/configs/config_template.py

class MinioConfig(BaseModel):
    """Configuration settings for MinIO/S3 object storage integration.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for Keycloak integration.

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

Source code in archipy/configs/config_template.py

class KeycloakConfig(BaseModel):
    """Configuration settings for Keycloak integration.

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

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

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

SERVER_URL: str | None = None

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

CLIENT_ID: str | None = None

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

REALM_NAME: str = 'master'

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

CLIENT_SECRET_KEY: str | None = None

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

VERIFY_SSL: bool = True

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

TIMEOUT: int = 10

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

IS_ADMIN_MODE_ENABLED: bool = False

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

ADMIN_USERNAME: str | None = None

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

ADMIN_PASSWORD: str | None = None

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

ADMIN_REALM_NAME: str = 'master'

options: show_root_toc_entry: false heading_level: 3

Web Framework Configs

Bases: BaseModel

Configuration settings for FastAPI applications.

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

Source code in archipy/configs/config_template.py

class FastAPIConfig(BaseModel):
    """Configuration settings for FastAPI applications.

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

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

    ACCESS_LOG: bool = Field(default=True, description="Whether to enable access logging")
    BACKLOG: int = Field(default=2048, description="Maximum number of queued connections")
    DATE_HEADER: bool = Field(default=True, description="Whether to include date header in responses")
    FORWARDED_ALLOW_IPS: list[str] | None = Field(default=None, description="List of allowed forwarded IPs")
    LIMIT_CONCURRENCY: int | None = Field(default=None, description="Maximum concurrent requests")
    LIMIT_MAX_REQUESTS: int | None = Field(default=None, description="Maximum number of requests")
    CORS_MIDDLEWARE_ALLOW_CREDENTIALS: bool = Field(default=True, description="Whether to allow credentials in CORS")
    CORS_MIDDLEWARE_ALLOW_HEADERS: list[str] = Field(default=["*"], description="Allowed CORS headers")
    CORS_MIDDLEWARE_ALLOW_METHODS: list[str] = Field(default=["*"], description="Allowed CORS methods")
    CORS_MIDDLEWARE_ALLOW_ORIGIN_REGEX: str | None = Field(
        default=None,
        description="Regex pattern for allowed origins",
    )
    CORS_MIDDLEWARE_ALLOW_ORIGINS: list[str] = Field(default=["*"], description="Allowed CORS origins")
    CORS_MIDDLEWARE_EXPOSE_HEADERS: list[str] = Field(default=[], description="Exposed CORS headers")
    CORS_MIDDLEWARE_MAX_AGE: int = Field(default=600, description="Preflight cache duration in seconds")
    GZIP_MIDDLEWARE_IS_ENABLED: bool = Field(default=False, description="Whether GZip response compression is enabled")
    GZIP_MIDDLEWARE_MINIMUM_SIZE: int = Field(
        default=500,
        ge=0,
        description="Minimum response body size in bytes before compression is applied",
    )
    GZIP_MIDDLEWARE_COMPRESSLEVEL: int = Field(
        default=6,
        ge=1,
        le=9,
        description="GZip compression level (1=fastest, 9=best compression)",
    )
    TRUSTED_HOST_MIDDLEWARE_IS_ENABLED: bool = Field(
        default=False,
        description="Whether TrustedHost middleware validates the Host header",
    )
    TRUSTED_HOST_MIDDLEWARE_ALLOWED_HOSTS: list[str] = Field(
        default=[],
        description="Allowed hostnames (e.g. example.com, *.example.com)",
    )
    TRUSTED_HOST_MIDDLEWARE_WWW_REDIRECT: bool = Field(
        default=True,
        description="Redirect www hosts to bare domain when allowed",
    )
    HTTPS_REDIRECT_MIDDLEWARE_IS_ENABLED: bool = Field(
        default=False,
        description="Whether HTTP requests are redirected to HTTPS",
    )
    PROXY_HEADERS: bool = Field(default=True, description="Whether to trust proxy headers")
    RELOAD: bool = Field(default=False, description="Whether to enable auto-reload")
    SERVER_HEADER: bool = Field(default=True, description="Whether to include server header")
    SERVE_HOST: str = Field(
        default="127.0.0.1",
        description="Host to serve the application on (set to 0.0.0.0 for container/public binding)",
    )
    SERVE_PORT: int = Field(default=8100, description="Port to serve the application on")
    TIMEOUT_GRACEFUL_SHUTDOWN: int | None = Field(default=None, description="Graceful shutdown timeout")
    TIMEOUT_KEEP_ALIVE: int = Field(default=5, description="Keep-alive timeout")
    WORKERS_COUNT: int = Field(default=4, description="Number of worker processes")
    WS_MAX_SIZE: int = Field(default=16777216, description="Maximum WebSocket message size")
    WS_PER_MESSAGE_DEFLATE: bool = Field(default=True, description="Whether to enable WebSocket compression")
    WS_PING_INTERVAL: float = Field(default=20.0, description="WebSocket ping interval")
    WS_PING_TIMEOUT: float = Field(default=20.0, description="WebSocket ping timeout")
    OPENAPI_URL: str | None = Field(default=None, description="URL for OpenAPI schema")
    DOCS_URL: str | None = Field(default=None, description="URL for API documentation")
    RE_DOC_URL: str | None = Field(default=None, description="URL for ReDoc documentation")
    SWAGGER_UI_PARAMS: dict[str, str] | None = Field(
        default={"docExpansion": "none"},
        description="Swagger UI parameters",
    )

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

SERVE_HOST: str = Field(
    default="127.0.0.1",
    description="Host to serve the application on (set to 0.0.0.0 for container/public binding)",
)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration for FastAPI REST rate limiting.

Controls trusted-proxy resolution, Redis key layout, failure modes, and standard rate-limit response headers for FastAPIRestRateLimitHandler.

Source code in archipy/configs/config_template.py

class FastAPIRateLimitConfig(BaseModel):
    """Configuration for FastAPI REST rate limiting.

    Controls trusted-proxy resolution, Redis key layout, failure modes, and
    standard rate-limit response headers for ``FastAPIRestRateLimitHandler``.
    """

    TRUSTED_PROXY_IPS: list[str] | None = Field(
        default=None,
        description=(
            "Trusted reverse-proxy CIDRs or IPs. When None, ``FASTAPI.FORWARDED_ALLOW_IPS`` is used. "
            "The literal ``*`` trusts all peers (matches Uvicorn; unsafe for production rate limiting)."
        ),
    )
    ALLOW_PRIVATE_IPS: bool = Field(
        default=False,
        description="Whether private, loopback, link-local, and multicast client addresses are accepted.",
    )
    KEY_PREFIX: str | None = Field(
        default=None,
        description="Redis key prefix. When None, ``{FASTAPI.PROJECT_NAME}:RateLimit`` is used.",
    )
    FAIL_CLOSED: bool = Field(
        default=True,
        description="When True, Redis or identity-resolution failures return HTTP 503 instead of allowing requests.",
    )
    SKIP_METHODS: list[str] = Field(
        default=["OPTIONS"],
        description="HTTP methods excluded from rate limiting.",
    )
    HASH_QUERY_PARAM_VALUES: bool = Field(
        default=True,
        description="Whether query parameter values are hashed in Redis keys to cap key length.",
    )
    REQUIRE_IDENTIFIER_FOR_QUERY_PARAMS: bool = Field(
        default=True,
        description=(
            "When True, ``query_params`` on the handler requires ``identifier_fn`` "
            "to avoid client-controlled per-user quota targeting."
        ),
    )
    REJECT_UNKNOWN_CLIENT: bool = Field(
        default=True,
        description="When True, requests without a resolvable client identity return HTTP 503.",
    )
    RATE_LIMIT_HEADERS: bool = Field(
        default=True,
        description="When True, attach X-RateLimit-* headers on allowed and rejected responses.",
    )
    IDENTITY_FROM_ACCESS_TOKEN: bool = Field(
        default=True,
        description=(
            "When True, verify Bearer access tokens via AUTH/JWTUtils and bucket by sub; "
            "fall back to client IP when the token is missing or invalid."
        ),
    )

archipy.configs.config_template.FastAPIRateLimitConfig.TRUSTED_PROXY_IPS class-attribute instance-attribute

TRUSTED_PROXY_IPS: list[str] | None = Field(
    default=None,
    description="Trusted reverse-proxy CIDRs or IPs. When None, ``FASTAPI.FORWARDED_ALLOW_IPS`` is used. The literal ``*`` trusts all peers (matches Uvicorn; unsafe for production rate limiting).",
)

archipy.configs.config_template.FastAPIRateLimitConfig.ALLOW_PRIVATE_IPS class-attribute instance-attribute

ALLOW_PRIVATE_IPS: bool = Field(
    default=False,
    description="Whether private, loopback, link-local, and multicast client addresses are accepted.",
)

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

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

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

FAIL_CLOSED: bool = Field(
    default=True,
    description="When True, Redis or identity-resolution failures return HTTP 503 instead of allowing requests.",
)

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

SKIP_METHODS: list[str] = Field(
    default=["OPTIONS"],
    description="HTTP methods excluded from rate limiting.",
)

archipy.configs.config_template.FastAPIRateLimitConfig.HASH_QUERY_PARAM_VALUES class-attribute instance-attribute

HASH_QUERY_PARAM_VALUES: bool = Field(
    default=True,
    description="Whether query parameter values are hashed in Redis keys to cap key length.",
)

archipy.configs.config_template.FastAPIRateLimitConfig.REQUIRE_IDENTIFIER_FOR_QUERY_PARAMS class-attribute instance-attribute

REQUIRE_IDENTIFIER_FOR_QUERY_PARAMS: bool = Field(
    default=True,
    description="When True, ``query_params`` on the handler requires ``identifier_fn`` to avoid client-controlled per-user quota targeting.",
)

archipy.configs.config_template.FastAPIRateLimitConfig.REJECT_UNKNOWN_CLIENT class-attribute instance-attribute

REJECT_UNKNOWN_CLIENT: bool = Field(
    default=True,
    description="When True, requests without a resolvable client identity return HTTP 503.",
)

archipy.configs.config_template.FastAPIRateLimitConfig.RATE_LIMIT_HEADERS class-attribute instance-attribute

RATE_LIMIT_HEADERS: bool = Field(
    default=True,
    description="When True, attach X-RateLimit-* headers on allowed and rejected responses.",
)

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

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

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration for gRPC server rate limiting.

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

Source code in archipy/configs/config_template.py

class GrpcRateLimitConfig(BaseModel):
    """Configuration for gRPC server rate limiting.

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

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

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

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

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

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

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

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

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

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

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

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

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for gRPC services.

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

Source code in archipy/configs/config_template.py

class GrpcConfig(BaseModel):
    """Configuration settings for gRPC services.

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

    SERVE_PORT: int = Field(default=8100, description="Port to serve gRPC on")
    SERVE_HOST: str = Field(default="[::]", description="Host to serve gRPC on")  # IPv6 equivalent of 0.0.0.0

    THREAD_WORKER_COUNT: int | None = Field(
        default=None,
        description=(
            "Number of worker threads. If omitted (None), set at model validation to "
            "THREAD_PER_CPU_CORE * (os.cpu_count() or 1)."
        ),
    )
    MAX_CONCURRENT_RPCS: int | None = Field(default=None, description="Maximum number of concurrent requests")
    THREAD_PER_CPU_CORE: int = Field(
        default=40,
        description="Threads per CPU core. Adjust based on thread block to CPU time ratio.",
    )

    SERVER_OPTIONS_CONFIG_LIST: list[tuple[str, int]] = Field(
        default=[
            # ── Message limits ──────────────────────────────────────────────
            ("grpc.max_metadata_size", 16 * 1024),  # 16KB  (default is 8KB)
            ("grpc.max_message_length", 128 * 1024 * 1024),  # 128MB
            ("grpc.max_receive_message_length", 128 * 1024 * 1024),
            ("grpc.max_send_message_length", 128 * 1024 * 1024),
            # ── Keepalive ───────────────────────────────────────────────────
            # Server rarely initiates pings; 2h is the standard default.
            ("grpc.keepalive_time_ms", 7_200_000),  # 2h
            ("grpc.keepalive_timeout_ms", 20_000),  # 20s
            ("grpc.keepalive_permit_without_calls", 1),  # allow pings on idle connections
            # ── Ping enforcement (server-side) ──────────────────────────────
            # Must be <= client keepalive_time_ms to avoid ENHANCE_YOUR_CALM.
            ("grpc.http2.min_recv_ping_interval_without_data_ms", 300_000),  # 5min
            ("grpc.http2.min_ping_interval_without_data_ms", 300_000),  # 5min
            ("grpc.http2.max_pings_without_data", 0),
            ("grpc.http2.max_ping_strikes", 2),
            # ── Connection lifetime ─────────────────────────────────────────
            ("grpc.max_connection_idle_ms", 600_000),  # 10min
            ("grpc.max_connection_age_ms", 1_800_000),  # 30min
            ("grpc.max_connection_age_grace_ms", 5_000),  # 5s graceful drain
        ],
        description="Server channel options",
    )

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

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

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

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

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

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

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

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

THREAD_WORKER_COUNT: int | None = Field(
    default=None,
    description="Number of worker threads. If omitted (None), set at model validation to THREAD_PER_CPU_CORE * (os.cpu_count() or 1).",
)

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

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

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

THREAD_PER_CPU_CORE: int = Field(
    default=40,
    description="Threads per CPU core. Adjust based on thread block to CPU time ratio.",
)

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

SERVER_OPTIONS_CONFIG_LIST: list[tuple[str, int]] = Field(
    default=[
        ("grpc.max_metadata_size", 16 * 1024),
        ("grpc.max_message_length", 128 * 1024 * 1024),
        (
            "grpc.max_receive_message_length",
            128 * 1024 * 1024,
        ),
        ("grpc.max_send_message_length", 128 * 1024 * 1024),
        ("grpc.keepalive_time_ms", 7200000),
        ("grpc.keepalive_timeout_ms", 20000),
        ("grpc.keepalive_permit_without_calls", 1),
        (
            "grpc.http2.min_recv_ping_interval_without_data_ms",
            300000,
        ),
        (
            "grpc.http2.min_ping_interval_without_data_ms",
            300000,
        ),
        ("grpc.http2.max_pings_without_data", 0),
        ("grpc.http2.max_ping_strikes", 2),
        ("grpc.max_connection_idle_ms", 600000),
        ("grpc.max_connection_age_ms", 1800000),
        ("grpc.max_connection_age_grace_ms", 5000),
    ],
    description="Server channel options",
)

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

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

archipy.configs.config_template.GrpcConfig.resolve_thread_worker_count

resolve_thread_worker_count() -> Self

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

Returns:

Type	Description
Self	This model with THREAD_WORKER_COUNT set when it was None.

Source code in archipy/configs/config_template.py

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

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

options: show_root_toc_entry: false heading_level: 3

Observability Configs

Bases: BaseModel

Configuration settings for 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")

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

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

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

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

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

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

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

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

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

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

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

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

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

SEND_DEFAULT_PII: bool = Field(
    default=False,
    description="Whether to send personally identifiable information (e.g. IP, user) to Sentry",
)

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

MAX_BREADCRUMBS: int = Field(
    default=100,
    description="Maximum number of breadcrumbs per event",
)

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

ATTACH_STACKTRACE: bool = Field(
    default=False,
    description="Whether to attach a stack trace to all messages (not only exceptions)",
)

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

SERVER_NAME: str | None = Field(
    default=None,
    description="Override the reported server hostname",
)

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

IN_APP_INCLUDE: list[str] = Field(
    default_factory=list,
    description="Module path prefixes to mark as in-app in stack traces",
)

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

IN_APP_EXCLUDE: list[str] = Field(
    default_factory=list,
    description="Module path prefixes to exclude from in-app marking",
)

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

PROFILES_SAMPLE_RATE: float = Field(
    default=0.0,
    description="Continuous profiling sample rate (0.0 to 1.0)",
)

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

IGNORE_ERRORS: list[str] = Field(
    default_factory=list,
    description="Exception type names (fully qualified or short) to ignore",
)

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

SHUTDOWN_TIMEOUT: int = Field(
    default=2,
    description="Seconds to wait for pending events on client shutdown",
)

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

DEFAULT_INTEGRATIONS: bool = Field(
    default=True,
    description="Whether to enable Sentry default integrations automatically",
)

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

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

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for Prometheus metrics integration.

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

Source code in archipy/configs/config_template.py

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

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

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

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

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

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

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

options: show_root_toc_entry: false heading_level: 3

Payment Configs

Bases: BaseModel

Configuration settings for Parsian Shaparak payment gateway integration.

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

Source code in archipy/configs/config_template.py

class ParsianShaparakConfig(BaseModel):
    """Configuration settings for Parsian Shaparak payment gateway integration.

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

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

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

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

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

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

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

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

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

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

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

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

options: show_root_toc_entry: false heading_level: 3

Application Configs

Bases: BaseModel

Configuration settings for authentication and security.

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

Source code in archipy/configs/config_template.py

class AuthConfig(BaseModel):
    """Configuration settings for authentication and security.

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

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

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

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

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

    # Token Security
    ENABLE_JTI_CLAIM: bool = Field(default=True, description="Enable JWT ID claim for token tracking")
    ENABLE_TOKEN_ROTATION: bool = Field(default=True, description="Enable refresh token rotation")
    REFRESH_TOKEN_REUSE_INTERVAL: int = Field(default=60, description="Grace period for refresh token reuse in seconds")

archipy.configs.config_template.AuthConfig.SECRET_KEY class-attribute instance-attribute

SECRET_KEY: SecretStr | None = Field(
    default=None, description="JWT signing key"
)

archipy.configs.config_template.AuthConfig.ACCESS_TOKEN_EXPIRES_IN class-attribute instance-attribute

ACCESS_TOKEN_EXPIRES_IN: int = Field(
    default=1 * 60 * 60,
    description="Access token expiration in seconds",
)

archipy.configs.config_template.AuthConfig.REFRESH_TOKEN_EXPIRES_IN class-attribute instance-attribute

REFRESH_TOKEN_EXPIRES_IN: int = Field(
    default=24 * 60 * 60,
    description="Refresh token expiration in seconds",
)

archipy.configs.config_template.AuthConfig.HASH_ALGORITHM class-attribute instance-attribute

HASH_ALGORITHM: str = Field(
    default="HS256", description="JWT signing algorithm"
)

archipy.configs.config_template.AuthConfig.JWT_ISSUER class-attribute instance-attribute

JWT_ISSUER: str = Field(
    default="your-app-name", description="JWT issuer claim"
)

archipy.configs.config_template.AuthConfig.JWT_AUDIENCE class-attribute instance-attribute

JWT_AUDIENCE: str = Field(
    default="your-app-audience",
    description="JWT audience claim",
)

archipy.configs.config_template.AuthConfig.TOKEN_VERSION class-attribute instance-attribute

TOKEN_VERSION: int = Field(
    default=1, description="JWT token version"
)

archipy.configs.config_template.AuthConfig.TOTP_SECRET_KEY class-attribute instance-attribute

TOTP_SECRET_KEY: SecretStr | None = Field(
    default=None, description="TOTP master key"
)

archipy.configs.config_template.AuthConfig.TOTP_HASH_ALGORITHM class-attribute instance-attribute

TOTP_HASH_ALGORITHM: str = Field(
    default="SHA1",
    description="Hash algorithm for TOTP generation (SHA1, SHA256, SHA512)",
)

archipy.configs.config_template.AuthConfig.TOTP_LENGTH class-attribute instance-attribute

TOTP_LENGTH: int = Field(
    default=6, ge=6, le=8, description="TOTP code length"
)

archipy.configs.config_template.AuthConfig.TOTP_EXPIRES_IN class-attribute instance-attribute

TOTP_EXPIRES_IN: int = Field(
    default=300,
    description="TOTP expiration time in seconds (5 minutes)",
)

archipy.configs.config_template.AuthConfig.TOTP_TIME_STEP class-attribute instance-attribute

TOTP_TIME_STEP: int = Field(
    default=30, description="TOTP time step in seconds"
)

archipy.configs.config_template.AuthConfig.TOTP_VERIFICATION_WINDOW class-attribute instance-attribute

TOTP_VERIFICATION_WINDOW: int = Field(
    default=1,
    description="Number of time steps to check before/after",
)

archipy.configs.config_template.AuthConfig.TOTP_MAX_ATTEMPTS class-attribute instance-attribute

TOTP_MAX_ATTEMPTS: int = Field(
    default=3,
    description="Maximum failed TOTP attempts before lockout",
)

archipy.configs.config_template.AuthConfig.TOTP_LOCKOUT_TIME class-attribute instance-attribute

TOTP_LOCKOUT_TIME: int = Field(
    default=300,
    description="Lockout time in seconds after max attempts",
)

archipy.configs.config_template.AuthConfig.LOGIN_RATE_LIMIT class-attribute instance-attribute

LOGIN_RATE_LIMIT: int = Field(
    default=5,
    description="Maximum login attempts per minute",
)

archipy.configs.config_template.AuthConfig.TOTP_RATE_LIMIT class-attribute instance-attribute

TOTP_RATE_LIMIT: int = Field(
    default=3,
    description="Maximum TOTP requests per minute",
)

archipy.configs.config_template.AuthConfig.PASSWORD_RESET_RATE_LIMIT class-attribute instance-attribute

PASSWORD_RESET_RATE_LIMIT: int = Field(
    default=3,
    description="Maximum password reset requests per hour",
)

archipy.configs.config_template.AuthConfig.HASH_ITERATIONS class-attribute instance-attribute

HASH_ITERATIONS: int = Field(
    default=100000, description="Password hash iterations"
)

archipy.configs.config_template.AuthConfig.MIN_LENGTH class-attribute instance-attribute

MIN_LENGTH: int = Field(
    default=12, ge=8, description="Minimum password length"
)

archipy.configs.config_template.AuthConfig.REQUIRE_DIGIT class-attribute instance-attribute

REQUIRE_DIGIT: bool = Field(
    default=True,
    description="Whether password requires digits",
)

archipy.configs.config_template.AuthConfig.REQUIRE_LOWERCASE class-attribute instance-attribute

REQUIRE_LOWERCASE: bool = Field(
    default=True,
    description="Whether password requires lowercase",
)

archipy.configs.config_template.AuthConfig.REQUIRE_SPECIAL class-attribute instance-attribute

REQUIRE_SPECIAL: bool = Field(
    default=True,
    description="Whether password requires special chars",
)

archipy.configs.config_template.AuthConfig.REQUIRE_UPPERCASE class-attribute instance-attribute

REQUIRE_UPPERCASE: bool = Field(
    default=True,
    description="Whether password requires uppercase",
)

archipy.configs.config_template.AuthConfig.SALT_LENGTH class-attribute instance-attribute

SALT_LENGTH: int = Field(
    default=16, description="Password salt length"
)

archipy.configs.config_template.AuthConfig.SPECIAL_CHARACTERS class-attribute instance-attribute

SPECIAL_CHARACTERS: set[str] = Field(
    default=set("!@#$%^&*()-_+="),
    description="Set of allowed special characters",
)

archipy.configs.config_template.AuthConfig.PASSWORD_HISTORY_SIZE class-attribute instance-attribute

PASSWORD_HISTORY_SIZE: int = Field(
    default=3,
    description="Number of previous passwords to remember",
)

archipy.configs.config_template.AuthConfig.ENABLE_JTI_CLAIM class-attribute instance-attribute

ENABLE_JTI_CLAIM: bool = Field(
    default=True,
    description="Enable JWT ID claim for token tracking",
)

archipy.configs.config_template.AuthConfig.ENABLE_TOKEN_ROTATION class-attribute instance-attribute

ENABLE_TOKEN_ROTATION: bool = Field(
    default=True,
    description="Enable refresh token rotation",
)

archipy.configs.config_template.AuthConfig.REFRESH_TOKEN_REUSE_INTERVAL class-attribute instance-attribute

REFRESH_TOKEN_REUSE_INTERVAL: int = Field(
    default=60,
    description="Grace period for refresh token reuse in seconds",
)

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for file handling capabilities.

Controls file link security, expiration policies, and file type restrictions.

Source code in archipy/configs/config_template.py

class FileConfig(BaseModel):
    """Configuration settings for file handling capabilities.

    Controls file link security, expiration policies, and file type restrictions.
    """

    SECRET_KEY: str | None = Field(default=None, description="Secret key used for generating secure file links")
    DEFAULT_EXPIRY_MINUTES: int = Field(
        default=60,
        ge=1,
        description="Default number of minutes until link expiration",  # Default 60 minutes (1 hour)
    )
    ALLOWED_EXTENSIONS: list[str] = Field(default=["jpg", "jpeg", "png"], description="List of allowed file extensions")

archipy.configs.config_template.FileConfig.SECRET_KEY class-attribute instance-attribute

SECRET_KEY: str | None = Field(
    default=None,
    description="Secret key used for generating secure file links",
)

archipy.configs.config_template.FileConfig.DEFAULT_EXPIRY_MINUTES class-attribute instance-attribute

DEFAULT_EXPIRY_MINUTES: int = Field(
    default=60,
    ge=1,
    description="Default number of minutes until link expiration",
)

archipy.configs.config_template.FileConfig.ALLOWED_EXTENSIONS class-attribute instance-attribute

ALLOWED_EXTENSIONS: list[str] = Field(
    default=["jpg", "jpeg", "png"],
    description="List of allowed file extensions",
)

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for date and time handling.

Controls API connections for specialized date/time services and date caching behavior.

Source code in archipy/configs/config_template.py

class DatetimeConfig(BaseModel):
    """Configuration settings for date and time handling.

    Controls API connections for specialized date/time services
    and date caching behavior.
    """

    TIME_IR_API_KEY: str | None = Field(
        default="ZAVdqwuySASubByCed5KYuYMzb9uB2f7",
        description="API key for time.ir service",
    )
    TIME_IR_API_ENDPOINT: str | None = Field(
        default="https://api.time.ir/v1/event/fa/events/calendar",
        description="Endpoint for time.ir service",
    )
    REQUEST_TIMEOUT: int = Field(default=5, description="Request timeout in seconds")
    MAX_RETRIES: int = Field(default=3, description="Maximum retry attempts")
    CACHE_TTL: int = Field(default=86400, description="Cache time-to-live in seconds (24 hours)")
    HISTORICAL_CACHE_TTL: int = Field(
        default=604800,
        description="Cache time-to-live for historical dates in seconds (7 days)",
    )

archipy.configs.config_template.DatetimeConfig.TIME_IR_API_KEY class-attribute instance-attribute

TIME_IR_API_KEY: str | None = Field(
    default="ZAVdqwuySASubByCed5KYuYMzb9uB2f7",
    description="API key for time.ir service",
)

archipy.configs.config_template.DatetimeConfig.TIME_IR_API_ENDPOINT class-attribute instance-attribute

TIME_IR_API_ENDPOINT: str | None = Field(
    default="https://api.time.ir/v1/event/fa/events/calendar",
    description="Endpoint for time.ir service",
)

archipy.configs.config_template.DatetimeConfig.REQUEST_TIMEOUT class-attribute instance-attribute

REQUEST_TIMEOUT: int = Field(
    default=5, description="Request timeout in seconds"
)

archipy.configs.config_template.DatetimeConfig.MAX_RETRIES class-attribute instance-attribute

MAX_RETRIES: int = Field(
    default=3, description="Maximum retry attempts"
)

archipy.configs.config_template.DatetimeConfig.CACHE_TTL class-attribute instance-attribute

CACHE_TTL: int = Field(
    default=86400,
    description="Cache time-to-live in seconds (24 hours)",
)

archipy.configs.config_template.DatetimeConfig.HISTORICAL_CACHE_TTL class-attribute instance-attribute

HISTORICAL_CACHE_TTL: int = Field(
    default=604800,
    description="Cache time-to-live for historical dates in seconds (7 days)",
)

options: show_root_toc_entry: false heading_level: 3

Bases: BaseModel

Configuration settings for Temporal workflow engine integration.

Controls connection parameters, security settings, and timeout configurations for Temporal workflow orchestration services.

Source code in archipy/configs/config_template.py

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)",
    )

    # Client Connection Configuration
    CLIENT_IDENTITY: str | None = Field(
        default=None,
        description="Client identity string for Temporal server visibility",
    )
    API_KEY: str | None = Field(
        default=None,
        description="API key for Temporal Cloud authentication (ignored by local dev server)",
    )
    LAZY_CONNECT: bool = Field(
        default=False,
        description="Defer establishing the Temporal client connection until first use",
    )
    KEEP_ALIVE_INTERVAL_MS: int = Field(
        default=30000,
        ge=1,
        description="gRPC keep-alive ping interval in milliseconds",
    )
    KEEP_ALIVE_TIMEOUT_MS: int = Field(
        default=15000,
        ge=1,
        description="gRPC keep-alive timeout in milliseconds",
    )
    RPC_METADATA: dict[str, str] | None = Field(
        default=None,
        description="Custom RPC metadata headers sent with every client request",
    )
    CLIENT_RPC_RETRY_MAX_RETRIES: int = Field(
        default=10,
        ge=0,
        description="Maximum client-side RPC retry attempts",
    )
    CLIENT_RPC_RETRY_INITIAL_INTERVAL_MS: int = Field(
        default=100,
        ge=1,
        description="Initial backoff interval for client-side RPC retries in milliseconds",
    )
    CLIENT_RPC_RETRY_MAX_INTERVAL_MS: int = Field(
        default=5000,
        ge=1,
        description="Maximum backoff interval for client-side RPC retries in milliseconds",
    )

    # Worker Configuration
    WORKER_GRACEFUL_SHUTDOWN_SECONDS: int = Field(
        default=30,
        ge=0,
        description="Maximum time in seconds to wait for graceful worker shutdown",
    )
    WORKER_MAX_CACHED_WORKFLOWS: int = Field(
        default=1000,
        ge=0,
        description="Maximum number of workflow instances cached in worker memory",
    )
    WORKER_DEBUG_MODE: bool = Field(
        default=False,
        description="Enable worker debug mode (e.g. breakpoint support in workflow sandbox)",
    )
    WORKER_MAX_CONCURRENT_LOCAL_ACTIVITIES: int | None = Field(
        default=None,
        ge=1,
        description="Maximum concurrent local activities (None uses server default)",
    )
    WORKER_DISABLE_EAGER_ACTIVITY_EXECUTION: bool = Field(
        default=False,
        description="Disable eager activity execution on the worker",
    )

    # TLS Configuration
    TLS_CA_CERT: str | None = Field(default=None, description="Path to TLS CA certificate")
    TLS_CLIENT_CERT: str | None = Field(default=None, description="Path to TLS client certificate")
    TLS_CLIENT_KEY: str | None = Field(default=None, description="Path to TLS client private key")

    # Workflow Timeout Settings
    WORKFLOW_EXECUTION_TIMEOUT: int = Field(
        default=300,
        ge=1,
        description="Maximum workflow execution time in seconds",
    )
    WORKFLOW_RUN_TIMEOUT: int = Field(
        default=60,
        ge=1,
        description="Maximum single workflow run time in seconds",
    )
    WORKFLOW_TASK_TIMEOUT: int = Field(
        default=30,
        ge=1,
        description="Maximum workflow task processing time in seconds",
    )

    # Activity Timeout Settings
    ACTIVITY_START_TO_CLOSE_TIMEOUT: int = Field(
        default=30,
        ge=1,
        description="Maximum activity execution time in seconds",
    )
    ACTIVITY_HEARTBEAT_TIMEOUT: int = Field(
        default=10,
        ge=1,
        description="Activity heartbeat timeout in seconds",
    )

    # Retry Configuration
    RETRY_MAXIMUM_ATTEMPTS: int = Field(
        default=3,
        ge=1,
        description="Maximum retry attempts for failed activities",
    )
    RETRY_BACKOFF_COEFFICIENT: float = Field(
        default=2.0,
        ge=1.0,
        description="Backoff multiplier for retry delays",
    )
    RETRY_MAXIMUM_INTERVAL: int = Field(
        default=60,
        ge=1,
        description="Maximum retry interval in seconds",
    )
    RETRY_INITIAL_INTERVAL: int = Field(
        default=1,
        ge=1,
        description="Initial retry interval in seconds for workflow and activity retry policies",
    )
    RETRY_NON_RETRYABLE_ERROR_TYPES: list[str] | None = Field(
        default=None,
        description="Error type names that should not be retried by workflow/activity retry policies",
    )

    @model_validator(mode="after")
    def validate_tls_configuration(self) -> Self:
        """Validate TLS configuration consistency."""
        tls_fields = [self.TLS_CA_CERT, self.TLS_CLIENT_CERT, self.TLS_CLIENT_KEY]
        tls_provided = [field for field in tls_fields if field is not None]

        if len(tls_provided) > 0 and len(tls_provided) != 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

    @model_validator(mode="after")
    def validate_keep_alive_configuration(self) -> Self:
        """Validate keep-alive timeout is less than the keep-alive interval."""
        if self.KEEP_ALIVE_TIMEOUT_MS >= self.KEEP_ALIVE_INTERVAL_MS:
            raise InvalidArgumentError()

        return self

    @model_validator(mode="after")
    def validate_rpc_metadata(self) -> Self:
        """Validate RPC metadata keys and values are non-empty strings."""
        if self.RPC_METADATA is None:
            return self

        for key, value in self.RPC_METADATA.items():
            if not key or not value:
                raise InvalidArgumentError()

        return self

archipy.configs.config_template.TemporalConfig.HOST class-attribute instance-attribute

HOST: str = Field(
    default="localhost",
    description="Temporal server host address",
)

archipy.configs.config_template.TemporalConfig.PORT class-attribute instance-attribute

PORT: int = Field(
    default=7233,
    ge=1,
    le=65535,
    description="Temporal server port number",
)

archipy.configs.config_template.TemporalConfig.NAMESPACE class-attribute instance-attribute

NAMESPACE: str = Field(
    default="default",
    description="Temporal namespace for workflow isolation",
)

archipy.configs.config_template.TemporalConfig.TASK_QUEUE class-attribute instance-attribute

TASK_QUEUE: str = Field(
    default="task-queue",
    description="Default task queue name",
)

archipy.configs.config_template.TemporalConfig.ENABLE_METRICS class-attribute instance-attribute

ENABLE_METRICS: bool = Field(
    default=False,
    description="Enable Prometheus metrics collection for Temporal workflows and activities",
)

archipy.configs.config_template.TemporalConfig.METRICS_PORT class-attribute instance-attribute

METRICS_PORT: int = Field(
    default=8201,
    ge=1,
    le=65535,
    description="Port for Temporal Prometheus metrics endpoint (separate from main Prometheus port)",
)

archipy.configs.config_template.TemporalConfig.CLIENT_IDENTITY class-attribute instance-attribute

CLIENT_IDENTITY: str | None = Field(
    default=None,
    description="Client identity string for Temporal server visibility",
)

archipy.configs.config_template.TemporalConfig.API_KEY class-attribute instance-attribute

API_KEY: str | None = Field(
    default=None,
    description="API key for Temporal Cloud authentication (ignored by local dev server)",
)

archipy.configs.config_template.TemporalConfig.LAZY_CONNECT class-attribute instance-attribute

LAZY_CONNECT: bool = Field(
    default=False,
    description="Defer establishing the Temporal client connection until first use",
)

archipy.configs.config_template.TemporalConfig.KEEP_ALIVE_INTERVAL_MS class-attribute instance-attribute

KEEP_ALIVE_INTERVAL_MS: int = Field(
    default=30000,
    ge=1,
    description="gRPC keep-alive ping interval in milliseconds",
)

archipy.configs.config_template.TemporalConfig.KEEP_ALIVE_TIMEOUT_MS class-attribute instance-attribute

KEEP_ALIVE_TIMEOUT_MS: int = Field(
    default=15000,
    ge=1,
    description="gRPC keep-alive timeout in milliseconds",
)

archipy.configs.config_template.TemporalConfig.RPC_METADATA class-attribute instance-attribute

RPC_METADATA: dict[str, str] | None = Field(
    default=None,
    description="Custom RPC metadata headers sent with every client request",
)

archipy.configs.config_template.TemporalConfig.CLIENT_RPC_RETRY_MAX_RETRIES class-attribute instance-attribute

CLIENT_RPC_RETRY_MAX_RETRIES: int = Field(
    default=10,
    ge=0,
    description="Maximum client-side RPC retry attempts",
)

archipy.configs.config_template.TemporalConfig.CLIENT_RPC_RETRY_INITIAL_INTERVAL_MS class-attribute instance-attribute

CLIENT_RPC_RETRY_INITIAL_INTERVAL_MS: int = Field(
    default=100,
    ge=1,
    description="Initial backoff interval for client-side RPC retries in milliseconds",
)

archipy.configs.config_template.TemporalConfig.CLIENT_RPC_RETRY_MAX_INTERVAL_MS class-attribute instance-attribute

CLIENT_RPC_RETRY_MAX_INTERVAL_MS: int = Field(
    default=5000,
    ge=1,
    description="Maximum backoff interval for client-side RPC retries in milliseconds",
)

archipy.configs.config_template.TemporalConfig.WORKER_GRACEFUL_SHUTDOWN_SECONDS class-attribute instance-attribute

WORKER_GRACEFUL_SHUTDOWN_SECONDS: int = Field(
    default=30,
    ge=0,
    description="Maximum time in seconds to wait for graceful worker shutdown",
)

archipy.configs.config_template.TemporalConfig.WORKER_MAX_CACHED_WORKFLOWS class-attribute instance-attribute

WORKER_MAX_CACHED_WORKFLOWS: int = Field(
    default=1000,
    ge=0,
    description="Maximum number of workflow instances cached in worker memory",
)

archipy.configs.config_template.TemporalConfig.WORKER_DEBUG_MODE class-attribute instance-attribute

WORKER_DEBUG_MODE: bool = Field(
    default=False,
    description="Enable worker debug mode (e.g. breakpoint support in workflow sandbox)",
)

archipy.configs.config_template.TemporalConfig.WORKER_MAX_CONCURRENT_LOCAL_ACTIVITIES class-attribute instance-attribute

WORKER_MAX_CONCURRENT_LOCAL_ACTIVITIES: int | None = Field(
    default=None,
    ge=1,
    description="Maximum concurrent local activities (None uses server default)",
)

archipy.configs.config_template.TemporalConfig.WORKER_DISABLE_EAGER_ACTIVITY_EXECUTION class-attribute instance-attribute

WORKER_DISABLE_EAGER_ACTIVITY_EXECUTION: bool = Field(
    default=False,
    description="Disable eager activity execution on the worker",
)

archipy.configs.config_template.TemporalConfig.TLS_CA_CERT class-attribute instance-attribute

TLS_CA_CERT: str | None = Field(
    default=None, description="Path to TLS CA certificate"
)

archipy.configs.config_template.TemporalConfig.TLS_CLIENT_CERT class-attribute instance-attribute

TLS_CLIENT_CERT: str | None = Field(
    default=None,
    description="Path to TLS client certificate",
)

archipy.configs.config_template.TemporalConfig.TLS_CLIENT_KEY class-attribute instance-attribute

TLS_CLIENT_KEY: str | None = Field(
    default=None,
    description="Path to TLS client private key",
)

archipy.configs.config_template.TemporalConfig.WORKFLOW_EXECUTION_TIMEOUT class-attribute instance-attribute

WORKFLOW_EXECUTION_TIMEOUT: int = Field(
    default=300,
    ge=1,
    description="Maximum workflow execution time in seconds",
)

archipy.configs.config_template.TemporalConfig.WORKFLOW_RUN_TIMEOUT class-attribute instance-attribute

WORKFLOW_RUN_TIMEOUT: int = Field(
    default=60,
    ge=1,
    description="Maximum single workflow run time in seconds",
)

archipy.configs.config_template.TemporalConfig.WORKFLOW_TASK_TIMEOUT class-attribute instance-attribute

WORKFLOW_TASK_TIMEOUT: int = Field(
    default=30,
    ge=1,
    description="Maximum workflow task processing time in seconds",
)

archipy.configs.config_template.TemporalConfig.ACTIVITY_START_TO_CLOSE_TIMEOUT class-attribute instance-attribute

ACTIVITY_START_TO_CLOSE_TIMEOUT: int = Field(
    default=30,
    ge=1,
    description="Maximum activity execution time in seconds",
)

archipy.configs.config_template.TemporalConfig.ACTIVITY_HEARTBEAT_TIMEOUT class-attribute instance-attribute

ACTIVITY_HEARTBEAT_TIMEOUT: int = Field(
    default=10,
    ge=1,
    description="Activity heartbeat timeout in seconds",
)

archipy.configs.config_template.TemporalConfig.RETRY_MAXIMUM_ATTEMPTS class-attribute instance-attribute

RETRY_MAXIMUM_ATTEMPTS: int = Field(
    default=3,
    ge=1,
    description="Maximum retry attempts for failed activities",
)

archipy.configs.config_template.TemporalConfig.RETRY_BACKOFF_COEFFICIENT class-attribute instance-attribute

RETRY_BACKOFF_COEFFICIENT: float = Field(
    default=2.0,
    ge=1.0,
    description="Backoff multiplier for retry delays",
)

archipy.configs.config_template.TemporalConfig.RETRY_MAXIMUM_INTERVAL class-attribute instance-attribute

RETRY_MAXIMUM_INTERVAL: int = Field(
    default=60,
    ge=1,
    description="Maximum retry interval in seconds",
)

archipy.configs.config_template.TemporalConfig.RETRY_INITIAL_INTERVAL class-attribute instance-attribute

RETRY_INITIAL_INTERVAL: int = Field(
    default=1,
    ge=1,
    description="Initial retry interval in seconds for workflow and activity retry policies",
)

archipy.configs.config_template.TemporalConfig.RETRY_NON_RETRYABLE_ERROR_TYPES class-attribute instance-attribute

RETRY_NON_RETRYABLE_ERROR_TYPES: list[str] | None = Field(
    default=None,
    description="Error type names that should not be retried by workflow/activity retry policies",
)

archipy.configs.config_template.TemporalConfig.validate_tls_configuration

validate_tls_configuration() -> Self

Validate TLS configuration consistency.

Source code in archipy/configs/config_template.py

@model_validator(mode="after")
def validate_tls_configuration(self) -> Self:
    """Validate TLS configuration consistency."""
    tls_fields = [self.TLS_CA_CERT, self.TLS_CLIENT_CERT, self.TLS_CLIENT_KEY]
    tls_provided = [field for field in tls_fields if field is not None]

    if len(tls_provided) > 0 and len(tls_provided) != 3:
        raise InvalidArgumentError()

    return self

archipy.configs.config_template.TemporalConfig.validate_timeout_hierarchy

validate_timeout_hierarchy() -> Self

Validate timeout configuration hierarchy.

Source code in archipy/configs/config_template.py

@model_validator(mode="after")
def validate_timeout_hierarchy(self) -> Self:
    """Validate timeout configuration hierarchy."""
    if self.WORKFLOW_RUN_TIMEOUT >= self.WORKFLOW_EXECUTION_TIMEOUT:
        raise InvalidArgumentError()

    if self.WORKFLOW_TASK_TIMEOUT >= self.WORKFLOW_RUN_TIMEOUT:
        raise InvalidArgumentError()

    return self

archipy.configs.config_template.TemporalConfig.validate_keep_alive_configuration

validate_keep_alive_configuration() -> Self

Validate keep-alive timeout is less than the keep-alive interval.

Source code in archipy/configs/config_template.py

@model_validator(mode="after")
def validate_keep_alive_configuration(self) -> Self:
    """Validate keep-alive timeout is less than the keep-alive interval."""
    if self.KEEP_ALIVE_TIMEOUT_MS >= self.KEEP_ALIVE_INTERVAL_MS:
        raise InvalidArgumentError()

    return self

archipy.configs.config_template.TemporalConfig.validate_rpc_metadata

validate_rpc_metadata() -> Self

Validate RPC metadata keys and values are non-empty strings.

Source code in archipy/configs/config_template.py

@model_validator(mode="after")
def validate_rpc_metadata(self) -> Self:
    """Validate RPC metadata keys and values are non-empty strings."""
    if self.RPC_METADATA is None:
        return self

    for key, value in self.RPC_METADATA.items():
        if not key or not value:
            raise InvalidArgumentError()

    return self

options: show_root_toc_entry: false heading_level: 3

Environment Type

archipy.configs.environment_type.EnvironmentType

Bases: StrEnum

Enum representing different application environments.

This enum defines the available environment types for an application and provides helper properties to check the current environment.

Attributes:

Name	Type	Description
PRODUCTION		Production environment
BETA		Beta testing environment (human testing)
ALPHA		Alpha testing environment (human testing)
TEST		Automated testing environment
DEV		Development environment
LOCAL		Local development environment

Examples:

>>> from archipy.configs.environment_type import EnvironmentType
>>>
>>> # Setting environment in configuration
>>> env = EnvironmentType.DEV
>>>
>>> # Checking environment type
>>> if env.is_production:
...     print("Running in production mode")
>>> elif env.is_dev:
...     print("Running in development mode")
>>>
>>> # Getting appropriate log level
>>> log_level = env.log_level
>>> print(f"Log level: {log_level}")

Source code in archipy/configs/environment_type.py

class EnvironmentType(StrEnum):
    """Enum representing different application environments.

    This enum defines the available environment types for an application
    and provides helper properties to check the current environment.

    Attributes:
        PRODUCTION: Production environment
        BETA: Beta testing environment (human testing)
        ALPHA: Alpha testing environment (human testing)
        TEST: Automated testing environment
        DEV: Development environment
        LOCAL: Local development environment

    Examples:
        >>> from archipy.configs.environment_type import EnvironmentType
        >>>
        >>> # Setting environment in configuration
        >>> env = EnvironmentType.DEV
        >>>
        >>> # Checking environment type
        >>> if env.is_production:
        ...     print("Running in production mode")
        >>> elif env.is_dev:
        ...     print("Running in development mode")
        >>>
        >>> # Getting appropriate log level
        >>> log_level = env.log_level
        >>> print(f"Log level: {log_level}")
    """

    PRODUCTION = "PRODUCTION"
    BETA = "BETA"  # human test
    ALPHA = "ALPHA"  # human test
    TEST = "TEST"  # automatic test
    DEV = "DEV"
    LOCAL = "LOCAL"

    @enum.property
    def is_local(self) -> bool:
        """Check if the environment is LOCAL.

        Returns:
            bool: True if environment is LOCAL, False otherwise.
        """
        return self == self.LOCAL

    @enum.property
    def is_dev(self) -> bool:
        """Check if the environment is DEV.

        Returns:
            bool: True if environment is DEV, False otherwise.
        """
        return self == self.DEV

    @enum.property
    def is_test(self) -> bool:
        """Check if the environment is a testing environment.

        Returns:
            bool: True if environment is BETA, ALPHA, or TEST, False otherwise.
        """
        return self in (self.BETA, self.ALPHA, self.TEST)

    @enum.property
    def is_production(self) -> bool:
        """Check if the environment is a production environment.

        This returns True for PRODUCTION and False for all other environments.

        Returns:
            bool: True if not a test, dev, or local environment, False otherwise.
        """
        return not self.is_test and not self.is_dev and not self.is_local

    @enum.property
    def log_level(self) -> int:
        """Get the appropriate logging level for this environment.

        Returns:
            int: WARNING for production, INFO for test environments,
                DEBUG for development and local environments.
        """
        if self.is_production:
            return WARNING
        if self.is_test:
            return INFO
        return DEBUG

archipy.configs.environment_type.EnvironmentType.PRODUCTION class-attribute instance-attribute

PRODUCTION = 'PRODUCTION'

archipy.configs.environment_type.EnvironmentType.BETA class-attribute instance-attribute

BETA = 'BETA'

archipy.configs.environment_type.EnvironmentType.ALPHA class-attribute instance-attribute

ALPHA = 'ALPHA'

archipy.configs.environment_type.EnvironmentType.TEST class-attribute instance-attribute

TEST = 'TEST'

archipy.configs.environment_type.EnvironmentType.DEV class-attribute instance-attribute

DEV = 'DEV'

archipy.configs.environment_type.EnvironmentType.LOCAL class-attribute instance-attribute

LOCAL = 'LOCAL'

archipy.configs.environment_type.EnvironmentType.is_local

is_local() -> bool

Check if the environment is LOCAL.

Returns:

Name	Type	Description
bool	bool	True if environment is LOCAL, False otherwise.

Source code in archipy/configs/environment_type.py

@enum.property
def is_local(self) -> bool:
    """Check if the environment is LOCAL.

    Returns:
        bool: True if environment is LOCAL, False otherwise.
    """
    return self == self.LOCAL

archipy.configs.environment_type.EnvironmentType.is_dev

is_dev() -> bool

Check if the environment is DEV.

Returns:

Name	Type	Description
bool	bool	True if environment is DEV, False otherwise.

Source code in archipy/configs/environment_type.py

@enum.property
def is_dev(self) -> bool:
    """Check if the environment is DEV.

    Returns:
        bool: True if environment is DEV, False otherwise.
    """
    return self == self.DEV

archipy.configs.environment_type.EnvironmentType.is_test

is_test() -> bool

Check if the environment is a testing environment.

Returns:

Name	Type	Description
bool	bool	True if environment is BETA, ALPHA, or TEST, False otherwise.

Source code in archipy/configs/environment_type.py

@enum.property
def is_test(self) -> bool:
    """Check if the environment is a testing environment.

    Returns:
        bool: True if environment is BETA, ALPHA, or TEST, False otherwise.
    """
    return self in (self.BETA, self.ALPHA, self.TEST)

archipy.configs.environment_type.EnvironmentType.is_production

is_production() -> bool

Check if the environment is a production environment.

This returns True for PRODUCTION and False for all other environments.

Returns:

Name	Type	Description
bool	bool	True if not a test, dev, or local environment, False otherwise.

Source code in archipy/configs/environment_type.py

@enum.property
def is_production(self) -> bool:
    """Check if the environment is a production environment.

    This returns True for PRODUCTION and False for all other environments.

    Returns:
        bool: True if not a test, dev, or local environment, False otherwise.
    """
    return not self.is_test and not self.is_dev and not self.is_local

archipy.configs.environment_type.EnvironmentType.log_level

log_level() -> int

Get the appropriate logging level for this environment.

Returns:

Name	Type	Description
int	int	WARNING for production, INFO for test environments, DEBUG for development and local environments.

Source code in archipy/configs/environment_type.py

@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