HartmutandClaude Sonnet 4.6
3c2d0816e5
H1: model_validator in Settings raises ValueError when jwt_secret_key is
"changeme" and the process is running inside a container — fast-fail
prevents insecure deployments; local dev outside Docker is unaffected.
H7: generate_order_number now acquires pg_advisory_xact_lock before SELECT MAX,
matching the same pattern used in generate_invoice_number (billing/service.py).
Concurrent order creation can no longer race to produce duplicate numbers.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
114 lines
3.5 KiB
Python
114 lines
3.5 KiB
Python
import os
|
|
from typing import Optional
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
from pydantic import model_validator
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
_DOCKER_SERVICE_ALIASES = {"postgres", "redis", "minio"}
|
|
|
|
|
|
def _is_running_in_container() -> bool:
|
|
if os.path.exists("/.dockerenv"):
|
|
return True
|
|
try:
|
|
with open("/proc/1/cgroup", "r", encoding="utf-8") as handle:
|
|
return "docker" in handle.read() or "containerd" in handle.read()
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _normalize_service_host(host: str) -> str:
|
|
if _is_running_in_container():
|
|
return host
|
|
return "localhost" if host in _DOCKER_SERVICE_ALIASES else host
|
|
|
|
|
|
def _normalize_service_url(url: str) -> str:
|
|
parsed = urlsplit(url)
|
|
if not parsed.hostname:
|
|
return url
|
|
|
|
normalized_host = _normalize_service_host(parsed.hostname)
|
|
if normalized_host == parsed.hostname:
|
|
return url
|
|
|
|
netloc = normalized_host
|
|
if parsed.username:
|
|
auth = parsed.username
|
|
if parsed.password:
|
|
auth = f"{auth}:{parsed.password}"
|
|
netloc = f"{auth}@{netloc}"
|
|
if parsed.port:
|
|
netloc = f"{netloc}:{parsed.port}"
|
|
return urlunsplit((parsed.scheme, netloc, parsed.path, parsed.query, parsed.fragment))
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
model_config = SettingsConfigDict(
|
|
env_file=".env",
|
|
case_sensitive=False,
|
|
extra="ignore",
|
|
)
|
|
|
|
# Database
|
|
postgres_db: str = "hartomat"
|
|
postgres_user: str = "hartomat"
|
|
postgres_password: str = "hartomat"
|
|
postgres_host: str = "localhost"
|
|
postgres_port: int = 5432
|
|
|
|
@property
|
|
def database_url(self) -> str:
|
|
return (
|
|
f"postgresql+asyncpg://{self.postgres_user}:{self.postgres_password}"
|
|
f"@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}"
|
|
)
|
|
|
|
@property
|
|
def database_url_sync(self) -> str:
|
|
return (
|
|
f"postgresql://{self.postgres_user}:{self.postgres_password}"
|
|
f"@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}"
|
|
)
|
|
|
|
# Redis / Celery
|
|
redis_url: str = "redis://localhost:6379/0"
|
|
workflow_shadow_render_queue: str = "asset_pipeline_light"
|
|
|
|
@model_validator(mode="after")
|
|
def normalize_runtime_hosts(self) -> "Settings":
|
|
self.postgres_host = _normalize_service_host(self.postgres_host)
|
|
self.redis_url = _normalize_service_url(self.redis_url)
|
|
return self
|
|
|
|
@model_validator(mode="after")
|
|
def reject_insecure_jwt_secret_in_production(self) -> "Settings":
|
|
if self.jwt_secret_key == "changeme" and _is_running_in_container():
|
|
raise ValueError(
|
|
"JWT_SECRET_KEY must be set to a secure random value in production. "
|
|
"The default 'changeme' key is not permitted when running inside a container."
|
|
)
|
|
return self
|
|
|
|
# JWT
|
|
jwt_secret_key: str = "changeme"
|
|
jwt_algorithm: str = "HS256"
|
|
jwt_access_token_expire_minutes: int = 480
|
|
|
|
# Azure OpenAI
|
|
azure_openai_api_key: Optional[str] = None
|
|
azure_openai_endpoint: Optional[str] = None
|
|
azure_openai_deployment: str = "gpt-4o"
|
|
azure_openai_api_version: str = "2024-02-01"
|
|
|
|
# Internal API (used by chat_service for self-calls — override in Docker if port changes)
|
|
internal_api_base_url: str = "http://localhost:8888"
|
|
|
|
# File Storage
|
|
upload_dir: str = "/app/uploads"
|
|
max_upload_size_mb: int = 500
|
|
|
|
settings = Settings()
|