shanxia/bx/bx-backend/app/services/base.py

from typing import TypeVar, Type, Sequence, Generic

from sqlalchemy.orm import Session

from app.models.base import BaseModel

M = TypeVar("M", bound=BaseModel)


class CrudService(Generic[M]):

    @staticmethod
    def get_by_id(model: Type[M], db: Session, id: int) -> M | None:
        query = db.query(model)
        if hasattr(model, "del_flag"):
            query = query.filter(model.del_flag == 0)
        return query.filter(model.id == id).first()

    @staticmethod
    def get_paged(
        model: Type[M],
        db: Session,
        page: int,
        page_size: int,
        filters: list | None = None,
    ) -> tuple[Sequence[M], int]:
        query = db.query(model)
        if hasattr(model, "del_flag"):
            query = query.filter(model.del_flag == 0)
        if filters:
            query = query.filter(*filters)
        total = query.count()
        # Default ordering: sort_order ASC, create_time DESC; if no sort_order, just create_time DESC
        if hasattr(model, "sort_order"):
            query = query.order_by(model.sort_order.asc(), model.create_time.desc())
        else:
            query = query.order_by(model.create_time.desc())
        rows = query.offset((page - 1) * page_size).limit(page_size).all()
        return rows, total

    @staticmethod
    def create(model: Type[M], db: Session, obj_in: dict) -> M:
        obj = model(**obj_in)
        db.add(obj)
        db.flush()
        return obj

    @staticmethod
    def update(model: Type[M], db: Session, id: int, obj_in: dict) -> M | None:
        query = db.query(model)
        if hasattr(model, "del_flag"):
            query = query.filter(model.del_flag == 0)
        obj = query.filter(model.id == id).first()
        if not obj:
            return None
        for k, v in obj_in.items():
            setattr(obj, k, v)
        db.flush()
        return obj

    @staticmethod
    def soft_delete(model: Type[M], db: Session, id: int) -> bool:
        query = db.query(model)
        if hasattr(model, "del_flag"):
            query = query.filter(model.del_flag == 0)
        obj = query.filter(model.id == id).first()
        if not obj:
            return False
        obj.del_flag = 1
        db.flush()
        return True