tmpl-fastapi/app/sql/db.py

from typing import AsyncGenerator

from pydantic import BaseSettings

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy_utils import database_exists
from sqlalchemy_utils import create_database


# Database configuration
class SqlSettings(BaseSettings):
    db_host: str = '${REPO_NAME_SNAKE}_db'
    db_port: int = 3306
    db_user: str = '${REPO_NAME_SNAKE}'
    db_password: str = ''
    db_database: str = '${REPO_NAME_SNAKE}'


sql_settings = SqlSettings()

# DB connection URL
# pylint: disable=consider-using-f-string
db_url = (
    'mysql://{db_user}:{db_password}@'
    '{db_host}:{db_port}/{db_database}'
).format(**sql_settings.dict())
# pylint: enable=consider-using-f-string


# SQLAlchemy engine object
engine = create_engine(db_url)

# Create DB if not exists
if not database_exists(db_url):
    create_database(db_url)

# SQLAlchemy Session object
SessionLocal = sessionmaker(
    autoflush=False,
    bind=engine,
)

# SQLAlchemy Base object
Base = declarative_base()


# FastAPI dependency
async def get_db() -> AsyncGenerator[Session, None]:
    """FastAPI dependency
    returning database Session object.
    Code is copied from the official docs

    Yields:
        SQLAlchemy Session object
    """

    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()