Files
X25020501_card_book/core/db/crud.py

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.insert_time))
)
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)