iTi-Flask/iti/config.py

from __future__ import annotations

import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from dotenv import load_dotenv


BASE_DIR = Path(os.getenv("ITI_BASE_DIR", Path.cwd())).resolve()


def load_env_file(env_dir: str | os.PathLike | None = None) -> bool:
    search_dir = Path(env_dir or os.getenv("ITI_ENV_DIR") or Path.cwd()).resolve()
    env_name = os.getenv("APP_ENV", os.getenv("ITI_ENV", "dev"))
    for name in (".env.local", f".env.{env_name}", ".env"):
        path = search_dir / name
        if path.exists():
            load_dotenv(path, override=False)
            return True
    return False


def env_bool(key: str, default: bool = False) -> bool:
    value = os.getenv(key)
    if value is None:
        return default
    return value.lower() in {"1", "true", "yes", "on"}


def default_mysql_url(database: str) -> str:
    return (
        f"mysql+pymysql://{os.getenv('MYSQL_USER', 'root')}:"
        f"{os.getenv('MYSQL_PASSWORD', 'password')}@"
        f"{os.getenv('MYSQL_HOST', '127.0.0.1')}:"
        f"{os.getenv('MYSQL_PORT', '3306')}/{database}?charset=utf8mb4"
    )


load_env_file()


@dataclass(slots=True)
class BaseConfig:
    app_name: str = "iTi"
    app_env: str = "dev"
    debug: bool = False
    testing: bool = False
    base_dir: Path = BASE_DIR

    secret_key: str = field(
        default_factory=lambda: os.getenv("SECRET_KEY", "dev-secret-key-change-me")
    )
    jwt_secret_key: str = field(
        default_factory=lambda: os.getenv("JWT_SECRET_KEY", "dev-jwt-secret-change-me")
    )
    jwt_algorithm: str = "HS256"
    jwt_access_token_expires_seconds: int = 3600
    jwt_refresh_token_expires_seconds: int = 30 * 24 * 3600

    database_url: str = field(
        default_factory=lambda: os.getenv(
            "DATABASE_URL", default_mysql_url(os.getenv("MYSQL_DATABASE", "iti_dev"))
        )
    )
    sqlalchemy_echo: bool = False
    sqlalchemy_pool_pre_ping: bool = True

    cors_origins: list[str] = field(default_factory=lambda: ["*"])

    health_enabled: bool = True
    ready_check_db: bool = False

    response_envelope_http_status: int = 200
    response_envelope_enabled: bool = True
    raw_response_paths: list[str] = field(
        default_factory=lambda: ["/health", "/ready", "/docs", "/openapi.json"]
    )
    docs_ui_enabled: list[str] = field(
        default_factory=lambda: ["swagger", "scalar", "redoc"]
    )
    output_camel_case: bool = True

    ratelimit_enabled: bool = True
    ratelimit_default: str = "1000 per hour"

    cache_enabled: bool = True
    cache_default_timeout: int = 300

    file_storage: dict[str, Any] = field(
        default_factory=lambda: {
            "DEFAULT_STORAGE_TYPE": "local",
            "MAX_FILE_SIZE": 100 * 1024 * 1024 * 1024,
            "TUS_CHUNK_SIZE": 5 * 1024 * 1024,
            "LOCAL": {
                "base_path": str(BASE_DIR / "runtime" / "uploads"),
            },
        }
    )

    services: dict[str, dict[str, Any]] = field(default_factory=dict)
    service_tokens: dict[str, str] = field(default_factory=dict)
    tasks_enabled: bool = False
    exchange_enabled: bool = True
    exchange_default_storage: str = "local"
    exchange_storage: dict[str, Any] = field(default_factory=dict)

    log_level: str = "INFO"
    log_dir: str = field(default_factory=lambda: str(BASE_DIR / "runtime" / "logs"))
    log_file_enabled: bool = False
    log_max_bytes: int = 50 * 1024 * 1024
    log_backup_count: int = 10
    log_json: bool = False

    audit_enabled: bool = False
    audit_service_name: str = "audit"
    audit_queue_size: int = 1000
    audit_batch_size: int = 20
    audit_flush_interval_seconds: float = 1.0


class DevConfig(BaseConfig):
    def __init__(self) -> None:
        super().__init__(
            app_env="dev",
            debug=True,
            database_url=os.getenv(
                "DATABASE_URL",
                default_mysql_url(os.getenv("MYSQL_DATABASE", "iti_dev")),
            ),
            sqlalchemy_echo=env_bool("SQLALCHEMY_ECHO", False),
            jwt_access_token_expires_seconds=24 * 3600,
            ratelimit_default="1000 per hour",
            log_file_enabled=env_bool("LOG_FILE_ENABLED", False),
        )


class TestConfig(BaseConfig):
    def __init__(self) -> None:
        super().__init__(
            app_env="test",
            testing=True,
            database_url=os.getenv(
                "DATABASE_URL",
                default_mysql_url(os.getenv("MYSQL_DATABASE", "iti_test")),
            ),
            ratelimit_enabled=False,
            log_file_enabled=False,
            audit_enabled=False,
        )


class ProdConfig(BaseConfig):
    def __init__(self) -> None:
        super().__init__(
            app_env="prod",
            debug=False,
            database_url=os.getenv(
                "DATABASE_URL",
                default_mysql_url(os.getenv("MYSQL_DATABASE", "iti_prod")),
            ),
            secret_key=os.getenv("SECRET_KEY", ""),
            jwt_secret_key=os.getenv("JWT_SECRET_KEY", ""),
            ratelimit_default="100 per hour",
            log_file_enabled=env_bool("LOG_FILE_ENABLED", True),
        )


config = {
    "dev": DevConfig,
    "test": TestConfig,
    "prod": ProdConfig,
    "default": DevConfig,
}


def get_config(env_name: str | None = None) -> BaseConfig:
    env_name = env_name or os.getenv("APP_ENV", os.getenv("ITI_ENV", "dev"))
    config_cls = config.get(env_name, config["default"])
    return config_cls()