167 lines
5.2 KiB
Python
167 lines
5.2 KiB
Python
"""CRUD基础类"""
|
|
from typing import TypeVar, Generic, Type, Optional, List, Any
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import and_, or_
|
|
import re
|
|
|
|
from app.core.utils import generate_id
|
|
from app.core.database import Base
|
|
from sqlalchemy import func
|
|
|
|
ModelType = TypeVar("ModelType", bound=Base)
|
|
|
|
|
|
def camel_to_snake(name: str) -> str:
|
|
"""驼峰命名转蛇形命名"""
|
|
s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
|
|
return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()
|
|
|
|
|
|
def get_model_attr(obj, key: str):
|
|
"""获取模型的属性名，支持驼峰转蛇形"""
|
|
# 先尝试直接匹配
|
|
if hasattr(obj, key):
|
|
return key
|
|
# 尝试驼峰转蛇形
|
|
snake_key = camel_to_snake(key)
|
|
if hasattr(obj, snake_key):
|
|
return snake_key
|
|
# 如果都没有，返回None
|
|
return None
|
|
|
|
|
|
class CRUDBase(Generic[ModelType]):
|
|
"""CRUD基类"""
|
|
|
|
def __init__(self, model: Type[ModelType]):
|
|
self.model = model
|
|
|
|
def get(self, db: Session, id: str) -> Optional[ModelType]:
|
|
"""根据ID获取单条记录"""
|
|
return db.query(self.model).filter(self.model.id == id).first()
|
|
|
|
def get_multi(
|
|
self,
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
filters: List[Any] = None,
|
|
order_by: str = None,
|
|
order_type: str = "desc"
|
|
) -> List[ModelType]:
|
|
"""获取多条记录"""
|
|
query = db.query(self.model)
|
|
if filters:
|
|
query = query.filter(*filters)
|
|
if order_by:
|
|
order_column = getattr(self.model, order_by, None)
|
|
if order_column:
|
|
if order_type == "asc":
|
|
query = query.order_by(order_column.asc())
|
|
else:
|
|
query = query.order_by(order_column.desc())
|
|
else:
|
|
query = query.order_by(self.model.create_at.desc())
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
def get_all(self, db: Session, filters: List[Any] = None) -> List[ModelType]:
|
|
"""获取所有记录"""
|
|
query = db.query(self.model)
|
|
if filters:
|
|
query = query.filter(*filters)
|
|
return query.all()
|
|
|
|
def create(self, db: Session, obj_in: dict, creator: dict = None) -> ModelType:
|
|
"""创建记录"""
|
|
obj = self.model()
|
|
for key, value in obj_in.items():
|
|
attr_name = get_model_attr(obj, key)
|
|
if attr_name:
|
|
setattr(obj, attr_name, value)
|
|
# 设置ID
|
|
if not obj.id:
|
|
obj.id = generate_id()
|
|
# 设置创建人信息
|
|
if creator:
|
|
if not obj.create_by and "user_id" in creator:
|
|
obj.create_by = creator["user_id"]
|
|
if not obj.create_name and "real_name" in creator:
|
|
obj.create_name = creator["real_name"]
|
|
db.add(obj)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return obj
|
|
|
|
def update(self, db: Session, id: str, obj_in: dict, updater: dict = None) -> Optional[ModelType]:
|
|
"""更新记录"""
|
|
obj = self.get(db, id)
|
|
if obj:
|
|
for key, value in obj_in.items():
|
|
attr_name = get_model_attr(obj, key)
|
|
if attr_name and key not in ["id", "create_at", "create_by", "create_name"]:
|
|
setattr(obj, attr_name, value)
|
|
# 设置更新人信息
|
|
if updater:
|
|
if "user_id" in updater:
|
|
obj.update_by = updater["user_id"]
|
|
if "real_name" in updater:
|
|
obj.update_name = updater.get("real_name", obj.create_name)
|
|
obj.update_at = func.now()
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return obj
|
|
|
|
def delete(self, db: Session, id: str) -> bool:
|
|
"""删除记录（软删除）"""
|
|
obj = self.get(db, id)
|
|
if obj:
|
|
if hasattr(obj, "is_del"):
|
|
obj.is_del = 1
|
|
else:
|
|
db.delete(obj)
|
|
db.commit()
|
|
return True
|
|
return False
|
|
|
|
def hard_delete(self, db: Session, id: str) -> bool:
|
|
"""物理删除"""
|
|
obj = self.get(db, id)
|
|
if obj:
|
|
db.delete(obj)
|
|
db.commit()
|
|
return True
|
|
return False
|
|
|
|
def count(self, db: Session, filters: List[Any] = None) -> int:
|
|
"""统计数量"""
|
|
query = db.query(self.model)
|
|
if filters:
|
|
query = query.filter(*filters)
|
|
return query.count()
|
|
|
|
def exists(self, db: Session, filters: List[Any] = None) -> bool:
|
|
"""检查记录是否存在"""
|
|
query = db.query(self.model)
|
|
if filters:
|
|
query = query.filter(*filters)
|
|
return query.first() is not None
|
|
|
|
|
|
class HistoryCRUD(CRUDBase[ModelType]):
|
|
"""历史表CRUD类"""
|
|
|
|
def create_history(self, db: Session, original_data: dict) -> ModelType:
|
|
"""创建历史记录"""
|
|
obj = self.model()
|
|
# 复制原始数据到历史表
|
|
for key, value in original_data.items():
|
|
if hasattr(obj, key):
|
|
setattr(obj, key, value)
|
|
# 设置历史记录创建时间
|
|
if hasattr(obj, "his_create_at"):
|
|
obj.his_create_at = func.now()
|
|
db.add(obj)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return obj
|