210 lines
7.2 KiB
Python
210 lines
7.2 KiB
Python
# -*- coding = utf-8 -*-
|
|
# @Time : 2025/2/13 下午12:18
|
|
# @File : crud.py
|
|
# @Software : PyCharm
|
|
# @Author : xingc
|
|
# @Desc :
|
|
from datetime import datetime
|
|
from typing import List, Optional, Union, NewType, Dict, Any, Tuple
|
|
|
|
from sqlalchemy import func, desc, delete, CursorResult
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.future import select
|
|
|
|
from .models import User, Card, SearchCardCache
|
|
|
|
Model = NewType('Model', Union[User, Card, SearchCardCache])
|
|
|
|
|
|
class CRUD:
|
|
def __init__(self, session: AsyncSession, model: Model):
|
|
self.session = session
|
|
self.model = model
|
|
|
|
async def get(self, filters: dict, model: Model = None) -> Optional[Model]:
|
|
model = model or self.model
|
|
|
|
result = await self.session.execute(select(model).filter(
|
|
*self.build_filter_conditions(filters, model))
|
|
)
|
|
record = result.scalar_one_or_none()
|
|
return record
|
|
|
|
async def get_paginated(
|
|
self, filters: dict, page: int = 1, page_size: int = 10, model: Model = None
|
|
) -> Tuple[List[Model], int]:
|
|
model = model or self.model
|
|
filter_conditions = self.build_filter_conditions(filters, model) if filters else []
|
|
query = select(model).filter(*filter_conditions).order_by(desc(model.created_at))
|
|
|
|
total_query = select(func.count()).select_from(model).filter(*filter_conditions)
|
|
total_result = await self.session.execute(total_query)
|
|
total = total_result.scalar()
|
|
|
|
query = query.limit(page_size).offset((page - 1) * page_size)
|
|
result = await self.session.execute(query)
|
|
records = result.scalars().all()
|
|
|
|
return records, total
|
|
|
|
async def get_grouped_count(
|
|
self, group_by_field: str, filters: Dict[str, Any] = None, model: Model = None
|
|
) -> Dict[Any, int]:
|
|
model = model or self.model
|
|
filter_conditions = self.build_filter_conditions(filters, model) if filters else []
|
|
group_by = getattr(model, group_by_field)
|
|
query = select(group_by, func.count()).filter(*filter_conditions).group_by(group_by)
|
|
|
|
result = await self.session.execute(query)
|
|
records = result.all()
|
|
|
|
return {record[0]: record[1] for record in records}
|
|
|
|
async def create(self, data: dict, filters: dict, model: Model = None) -> Union[Model, bool]:
|
|
model = model or self.model
|
|
|
|
result = await self.session.execute(select(model).filter(
|
|
*self.build_filter_conditions(filters, model))
|
|
)
|
|
if result.scalars().first() is not None:
|
|
return False
|
|
|
|
record = model(**data)
|
|
self.session.add(record)
|
|
await self.session.commit()
|
|
return record
|
|
|
|
async def get_all(self, filters: dict = None, model: Model = None) -> List[Model]:
|
|
model = model or self.model
|
|
|
|
result = await self.session.execute(
|
|
select(model) if filters is None else select(model).filter(
|
|
*self.build_filter_conditions(filters, model)).order_by(desc(model.created_at))
|
|
)
|
|
records = result.scalars().all()
|
|
return records
|
|
|
|
async def update(self, data: dict, filters: dict, model: Model = None) -> Optional[Model]:
|
|
model = model or self.model
|
|
|
|
result = await self.session.execute(select(model).filter(
|
|
*self.build_filter_conditions(filters, model))
|
|
)
|
|
record = result.scalars().first()
|
|
|
|
if not record:
|
|
return None
|
|
|
|
count = 0
|
|
for key, value in data.items():
|
|
if any([
|
|
value is None,
|
|
value == getattr(record, key),
|
|
key in ['created_at', 'updated_at']
|
|
]):
|
|
continue
|
|
setattr(record, key, value)
|
|
count += 1
|
|
if count > 0:
|
|
await self.session.commit()
|
|
return record
|
|
|
|
async def updates(self, data: dict, filters: dict, model: Model = None) -> Optional[Model]:
|
|
model = model or self.model
|
|
|
|
result = await self.session.execute(select(model).filter(
|
|
*self.build_filter_conditions(filters, model))
|
|
)
|
|
records = result.scalars().all()
|
|
|
|
if len(records) == 0:
|
|
return None
|
|
count = 0
|
|
for record in records:
|
|
for key, value in data.items():
|
|
if any([
|
|
value is None,
|
|
value == getattr(record, key),
|
|
key in ['insert_time', 'update_time']
|
|
]):
|
|
continue
|
|
setattr(record, key, value)
|
|
count += 1
|
|
if count > 0:
|
|
await self.session.commit()
|
|
return records
|
|
|
|
async def delete(self, filters: dict, model: Model = None) -> bool:
|
|
model = model or self.model
|
|
|
|
result = await self.session.execute(delete(model).filter(
|
|
*self.build_filter_conditions(filters, model))
|
|
)
|
|
await self.session.commit()
|
|
return result.rowcount > 0
|
|
|
|
async def deletes(self, filters: dict, model: Model = None) -> bool:
|
|
model = model or self.model
|
|
|
|
result = await self.session.execute(delete(model).filter(
|
|
*self.build_filter_conditions(filters, model))
|
|
)
|
|
await self.session.commit()
|
|
return result.rowcount > 0
|
|
|
|
async def exists(self, filters: dict, model: Model = None) -> bool:
|
|
model = model or self.model
|
|
|
|
result = await self.session.execute(select(model).filter(
|
|
*self.build_filter_conditions(filters, model))
|
|
)
|
|
state = True if result.scalars().first() else False
|
|
return state
|
|
|
|
async def total(self, filters: dict, model: Model = None) -> int:
|
|
model = model or self.model
|
|
|
|
result = await self.session.execute(select(func.count()).filter(
|
|
*self.build_filter_conditions(filters, model))
|
|
)
|
|
return result.scalar()
|
|
|
|
def build_filter_conditions(self, filters: Dict[str, Any], model: Model) -> tuple:
|
|
conditions = []
|
|
|
|
for key, value in filters.items():
|
|
if any([
|
|
value is None,
|
|
value == '',
|
|
]):
|
|
continue
|
|
if isinstance(value, datetime) and value.tzinfo is not None:
|
|
value = value.astimezone()
|
|
|
|
# 解析字段名和操作符
|
|
field_name, operator = key, None
|
|
if '__' in key:
|
|
field_name, operator = key.split('__', 1)
|
|
else:
|
|
field_name = key
|
|
|
|
field = getattr(model, field_name)
|
|
|
|
if operator == 'like': # 模糊匹配
|
|
conditions.append(field.like(f'%{value}%'))
|
|
elif operator == 'gt': # 大于
|
|
conditions.append(field > value)
|
|
elif operator == 'lt': # 小于
|
|
conditions.append(field < value)
|
|
elif operator == 'eq': # 等于
|
|
conditions.append(field == value)
|
|
elif operator == 'ne': # 不等于
|
|
conditions.append(field != value)
|
|
elif operator == 'in': # 在集合中
|
|
conditions.append(field.in_(value))
|
|
else:
|
|
# 如果没有匹配的操作符，默认为等于
|
|
conditions.append(field == value)
|
|
|
|
return tuple(conditions)
|