MotorSignal-App/backend/motorsignal_api/app/db.py

31 lines
711 B
Python

import os
from sqlalchemy import create_engine, text
from sqlalchemy.engine import URL
def database_url() -> URL:
return URL.create(
drivername="postgresql+psycopg",
username=os.environ["DATABASE_USER"],
password=os.environ["DATABASE_PASSWORD"],
host=os.environ.get("DATABASE_HOST", "127.0.0.1"),
port=int(os.environ.get("DATABASE_PORT", "5432")),
database=os.environ["DATABASE_NAME"],
)
engine = create_engine(
database_url(),
pool_pre_ping=True,
pool_size=5,
max_overflow=5,
pool_timeout=10,
)
def check_database() -> bool:
with engine.connect() as connection:
connection.execute(text("SELECT 1"))
return True