Skip to content

settings

Configuration management for Direktor using Pydantic Settings.

direktor.core.settings

Application settings for Direktor.

Uses Pydantic Settings to load configuration from environment variables and .env files with validation, type coercion, and lazy initialization.

Settings

Bases: BaseSettings

Centralized, validated settings for the Direktor pipeline.

Source code in direktor/core/settings.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class Settings(BaseSettings):
    """Centralized, validated settings for the Direktor pipeline."""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    # API tokens
    replicate_api_token: str
    openai_api_key: str

    # Model identifiers
    distil_model: str
    bark_model: str
    flux_model: str
    gpt4_model: str = "gpt-4o"
    gpt4_max_tokens: int = 8000

    # S3-compatible storage (R2/Backblaze/etc.)
    aws_access_key_id: str | None = None
    aws_secret_access_key: str | None = None
    aws_endpoint_url: str | None = None
    aws_bucket_name: str | None = None
    aws_region_name: str = "auto"

    # Processing behavior
    temp_dir_base: str = "temp"
    max_chunk_chars: int = 150
    target_segment_duration: int = 30

    # Logging
    log_level: str = "INFO"

    @field_validator("log_level")
    @classmethod
    def _validate_log_level(cls, value: str) -> str:
        allowed = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
        upper = value.upper()
        if upper not in allowed:
            raise ValueError(
                f"Invalid log level '{value}'. Must be one of: {', '.join(sorted(allowed))}"
            )
        return upper

    def required_env_vars(self) -> list[str]:
        """Return the list of required environment variable names."""
        return [
            "REPLICATE_API_TOKEN",
            "OPENAI_API_KEY",
            "DISTIL_MODEL",
            "BARK_MODEL",
            "FLUX_MODEL",
            "GPT4_MODEL",
        ]

    def ensure_valid(self) -> None:
        """Validate that all required settings are present."""
        missing = [
            name
            for name in self.required_env_vars()
            if not getattr(self, name.lower(), None)
        ]
        if missing:
            raise ValueError(
                f"Missing required environment variables: {', '.join(missing)}. "
                "Please set these in your .env file or environment."
            )

    @property
    def client(self) -> OpenAI:
        """Lazy OpenAI client initialized from the API key."""
        if self.openai_api_key is None:
            raise ValueError("OPENAI_API_KEY is not set.")
        return OpenAI(api_key=self.openai_api_key)

    @property
    def encoding(self) -> Encoding:
        """Lazy tiktoken encoder for the configured GPT model."""
        if self.gpt4_model is None:
            raise ValueError("GPT4_MODEL is not set.")
        return encoding_for_model(self.gpt4_model)

    @property
    def assets_dir(self) -> Path:
        """Path to the bundled assets directory."""
        return Path(__file__).resolve().parent.parent / "assets"

    @property
    def font_path(self) -> Path:
        """Path to the bundled title font."""
        return self.assets_dir / "mexcellent_3d.ttf"

    @property
    def replicate_token(self) -> str:
        """Alias for ``replicate_api_token`` used in module code."""
        return self.replicate_api_token

assets_dir property

Path to the bundled assets directory.

client property

Lazy OpenAI client initialized from the API key.

encoding property

Lazy tiktoken encoder for the configured GPT model.

font_path property

Path to the bundled title font.

replicate_token property

Alias for replicate_api_token used in module code.

ensure_valid()

Validate that all required settings are present.

Source code in direktor/core/settings.py 
78
79
80
81
82
83
84
85
86
87
88
89
def ensure_valid(self) -> None:
    """Validate that all required settings are present."""
    missing = [
        name
        for name in self.required_env_vars()
        if not getattr(self, name.lower(), None)
    ]
    if missing:
        raise ValueError(
            f"Missing required environment variables: {', '.join(missing)}. "
            "Please set these in your .env file or environment."
        )

required_env_vars()

Return the list of required environment variable names.

Source code in direktor/core/settings.py 
67
68
69
70
71
72
73
74
75
76
def required_env_vars(self) -> list[str]:
    """Return the list of required environment variable names."""
    return [
        "REPLICATE_API_TOKEN",
        "OPENAI_API_KEY",
        "DISTIL_MODEL",
        "BARK_MODEL",
        "FLUX_MODEL",
        "GPT4_MODEL",
    ]

get_settings() cached

Return the cached, validated application settings.

Source code in direktor/core/settings.py 
121
122
123
124
125
126
127
128
129
130
131
@lru_cache
def get_settings() -> Settings:
    """Return the cached, validated application settings."""
    # Pydantic Settings reads values from the environment / .env file, so the
    # apparent missing constructor arguments are resolved at runtime.
    settings: Settings = Settings()  # type: ignore[call-arg]
    settings.ensure_valid()
    # Ensure replicate library sees the token if it reads the env var directly.
    if settings.replicate_api_token:
        os.environ["REPLICATE_API_TOKEN"] = settings.replicate_api_token
    return settings

reset_settings()

Reset the cached settings instance (useful mainly in tests).

Source code in direktor/core/settings.py 
134
135
136
def reset_settings() -> None:
    """Reset the cached settings instance (useful mainly in tests)."""
    get_settings.cache_clear()