65 lines
1.7 KiB
Python
65 lines
1.7 KiB
Python
from sqlalchemy import create_engine
|
|
from sqlalchemy.engine import URL
|
|
from sqlalchemy.orm import declarative_base, sessionmaker
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
|
|
load_dotenv()
|
|
|
|
|
|
def _build_database_url() -> str:
|
|
"""Construct the SQLAlchemy database URL from granular environment vars.
|
|
|
|
Falls back to `DATABASE_URL` for backward compatibility.
|
|
"""
|
|
|
|
legacy_url = os.environ.get("DATABASE_URL")
|
|
if legacy_url:
|
|
return legacy_url
|
|
|
|
driver = os.environ.get("DATABASE_DRIVER", "postgresql")
|
|
host = os.environ.get("DATABASE_HOST")
|
|
port = os.environ.get("DATABASE_PORT", "5432")
|
|
user = os.environ.get("DATABASE_USER")
|
|
password = os.environ.get("DATABASE_PASSWORD")
|
|
database = os.environ.get("DATABASE_NAME")
|
|
schema = os.environ.get("DATABASE_SCHEMA", "public")
|
|
|
|
missing = [
|
|
var_name
|
|
for var_name, value in (
|
|
("DATABASE_HOST", host),
|
|
("DATABASE_USER", user),
|
|
("DATABASE_NAME", database),
|
|
)
|
|
if not value
|
|
]
|
|
|
|
if missing:
|
|
raise RuntimeError(
|
|
"Missing database configuration: set DATABASE_URL or provide "
|
|
f"granular variables ({', '.join(missing)})"
|
|
)
|
|
|
|
url = URL.create(
|
|
drivername=driver,
|
|
username=user,
|
|
password=password,
|
|
host=host,
|
|
port=int(port) if port else None,
|
|
database=database,
|
|
)
|
|
|
|
if schema:
|
|
url = url.set(query={"options": f"-csearch_path={schema}"})
|
|
|
|
return str(url)
|
|
|
|
|
|
DATABASE_URL = _build_database_url()
|
|
|
|
engine = create_engine(DATABASE_URL, echo=True, future=True)
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
Base = declarative_base()
|