vent_ai_platform/app/db/repositories/base.py

from __future__ import annotations

from typing import Any, TypeVar

from sqlalchemy import Select, func, select
from sqlalchemy.orm import Session

from app.db.base import Base


ModelT = TypeVar("ModelT", bound=Base)


def create_model(db: Session, model: type[ModelT], data: dict[str, Any]) -> ModelT:
    instance = model(**data)
    db.add(instance)
    db.commit()
    db.refresh(instance)
    return instance


def update_model(
    db: Session,
    instance: ModelT,
    data: dict[str, Any],
) -> ModelT:
    for key, value in data.items():
        setattr(instance, key, value)
    db.add(instance)
    db.commit()
    db.refresh(instance)
    return instance


def delete_model(db: Session, instance: ModelT) -> None:
    db.delete(instance)
    db.commit()


def paginate(
    db: Session,
    statement: Select,
    page: int,
    page_size: int,
) -> tuple[list[Any], int]:
    count_statement = select(func.count()).select_from(statement.order_by(None).subquery())
    total = int(db.scalar(count_statement) or 0)
    rows = db.scalars(
        statement.offset((page - 1) * page_size).limit(page_size)
    ).all()
    return list(rows), total