mirror of
https://github.com/Kludex/awesome-fastapi-projects.git
synced 2025-05-12 20:35:35 +00:00
108 lines
3.3 KiB
Python
108 lines
3.3 KiB
Python
"""
|
|
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 collections.abc import AsyncGenerator
|
|
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,
|
|
)
|
|
|
|
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",
|
|
}
|
|
)
|
|
|
|
|
|
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
|
|
"""Get an async session."""
|
|
async with async_session_maker() as session:
|
|
yield session
|
|
|
|
|
|
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[int | None] = mapped_column(
|
|
BigInteger, nullable=True, unique=True
|
|
)
|
|
dependencies: Mapped[list["Dependency"]] = relationship(
|
|
"Dependency", secondary="repo_dependency", back_populates="repos"
|
|
)
|
|
last_checked_revision: Mapped[str | 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
|
|
)
|