40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ServiceSettings:
|
|
db_dsn: str
|
|
|
|
@staticmethod
|
|
def from_env() -> "ServiceSettings":
|
|
dsn = (
|
|
os.getenv("CLUSTERING_DB_DSN")
|
|
or os.getenv("DATABASE_URL")
|
|
or _build_dsn_from_parts()
|
|
)
|
|
if not dsn:
|
|
raise RuntimeError(
|
|
"No database connection configured. Set CLUSTERING_DB_DSN or DATABASE_URL, "
|
|
"or provide CLUSTERING_DB_HOST / CLUSTERING_DB_PORT / CLUSTERING_DB_NAME / "
|
|
"CLUSTERING_DB_USER / CLUSTERING_DB_PASSWORD."
|
|
)
|
|
return ServiceSettings(db_dsn=dsn)
|
|
|
|
|
|
def _build_dsn_from_parts() -> str | None:
|
|
host = os.getenv("CLUSTERING_DB_HOST")
|
|
database = os.getenv("CLUSTERING_DB_NAME")
|
|
user = os.getenv("CLUSTERING_DB_USER")
|
|
password = os.getenv("CLUSTERING_DB_PASSWORD")
|
|
port = os.getenv("CLUSTERING_DB_PORT", "5432")
|
|
|
|
if not host or not database or not user:
|
|
return None
|
|
|
|
if password:
|
|
return f"postgresql://{user}:{password}@{host}:{port}/{database}"
|
|
return f"postgresql://{user}@{host}:{port}/{database}"
|