zwitschi
97b1c0360b
- Updated test functions in various test files to enhance code clarity by formatting long lines and improving indentation. - Adjusted assertions to use multi-line formatting for better readability. - Added new test cases for theme settings API to ensure proper functionality. - Ensured consistent use of line breaks and spacing across test files for uniformity.
67 lines
1.7 KiB
Python
67 lines
1.7 KiB
Python
from sqlalchemy import create_engine
|
|
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 and legacy_url.strip() != "":
|
|
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 = f"{driver}://{user}:{password}@{host}"
|
|
if port:
|
|
url += f":{port}"
|
|
url += f"/{database}"
|
|
if schema:
|
|
url += f"?options=-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()
|
|
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|