awesome-fastapi-projects/app/database.py

"""
Database models and session.

The database is a SQLite database, and is stored in the root
of the project as `db.sqlite3`.

The database is managed using Alembic, and migrations
are stored in the `migrations/` directory.

The module defines the following models:

- `Repo`: A repository that is being tracked.
- `Dependency`: A dependency of a repository.
- `RepoDependency`: A relationship between a repository and a dependency.

The database is accessed asynchronously using SQLAlchemy's async API.
"""
from pathlib import PurePath
from typing import Final

from sqlalchemy import BigInteger, ForeignKey, MetaData, String, Text, UniqueConstraint
from sqlalchemy.ext.asyncio import (
    AsyncAttrs,
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import (
    Mapped,
    declarative_base,
    mapped_column,
    relationship,
)

from app.types import RevisionHash, SourceGraphRepoId

_DB_PATH: Final[PurePath] = PurePath(__file__).parent.parent / "db.sqlite3"

_SQLALCHEMY_DATABASE_URL: Final[str] = f"sqlite+aiosqlite:///{_DB_PATH}"

engine: Final[AsyncEngine] = create_async_engine(_SQLALCHEMY_DATABASE_URL)

async_session_maker: Final[async_sessionmaker[AsyncSession]] = async_sessionmaker(
    engine, expire_on_commit=False, autoflush=False, autocommit=False
)

metadata = MetaData(
    naming_convention={
        "ix": "ix_%(table_name)s_%(column_0_N_name)s ",
        "uq": "uq_%(table_name)s_%(column_0_N_name)s ",
        "ck": "ck_%(table_name)s_%(constraint_name)s ",
        "fk": "fk_%(table_name)s_%(column_0_N_name)s_%(referred_table_name)s",
        "pk": "pk_%(table_name)s",
    }
)


Base = declarative_base(metadata=metadata, cls=AsyncAttrs)


class Repo(Base):
    """A repository that is being tracked."""

    __tablename__ = "repo"
    id: Mapped[int] = mapped_column(primary_key=True)
    url: Mapped[str] = mapped_column(nullable=False, unique=True)
    description: Mapped[str] = mapped_column(Text, nullable=False)
    stars: Mapped[int] = mapped_column(BigInteger, nullable=False)
    source_graph_repo_id: Mapped[SourceGraphRepoId | None] = mapped_column(
        BigInteger, nullable=True, unique=True
    )
    dependencies: Mapped[list["Dependency"]] = relationship(
        "Dependency", secondary="repo_dependency", back_populates="repos"
    )
    last_checked_revision: Mapped[RevisionHash | None] = mapped_column(
        String(255), nullable=True
    )
    __table_args__ = (UniqueConstraint("url", "source_graph_repo_id"),)


class Dependency(Base):
    """A dependency of a repository."""

    __tablename__ = "dependency"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
    repos: Mapped[list["Repo"]] = relationship(
        "Repo", secondary="repo_dependency", back_populates="dependencies"
    )


class RepoDependency(Base):
    """A relationship between a repository and a dependency."""

    __tablename__ = "repo_dependency"
    repo_id: Mapped[int] = mapped_column(
        ForeignKey(Repo.id, ondelete="CASCADE"), primary_key=True
    )
    dependency_id: Mapped[int] = mapped_column(
        ForeignKey(Dependency.id, ondelete="CASCADE"), primary_key=True
    )