feat(db): 调整统一管理事务方式
This commit is contained in:
xingc
160
core/db/crud.py
160
core/db/crud.py
@@ -23,28 +23,27 @@ class CRUD:
|
|||||||
|
|
||||||
async def get(self, filters: dict, model: Model = None) -> Optional[Model]:
|
async def get(self, filters: dict, model: Model = None) -> Optional[Model]:
|
||||||
model = model or self.model
|
model = model or self.model
|
||||||
async with self.session.begin():
|
|
||||||
result = await self.session.execute(select(model).filter(
|
result = await self.session.execute(select(model).filter(
|
||||||
*self.build_filter_conditions(filters, model))
|
*self.build_filter_conditions(filters, model))
|
||||||
)
|
)
|
||||||
record = result.scalar_one_or_none()
|
record = result.scalar_one_or_none()
|
||||||
return record
|
return record
|
||||||
|
|
||||||
async def get_paginated(
|
async def get_paginated(
|
||||||
self, filters: dict, page: int = 1, page_size: int = 10, model: Model = None
|
self, filters: dict, page: int = 1, page_size: int = 10, model: Model = None
|
||||||
) -> Tuple[List[Model], int]:
|
) -> Tuple[List[Model], int]:
|
||||||
model = model or self.model
|
model = model or self.model
|
||||||
async with self.session.begin():
|
filter_conditions = self.build_filter_conditions(filters, model) if filters else []
|
||||||
filter_conditions = self.build_filter_conditions(filters) if filters else []
|
query = select(model).filter(*filter_conditions).order_by(desc(model.created_at))
|
||||||
query = select(model).filter(*filter_conditions).order_by(desc(model.created_at))
|
|
||||||
|
|
||||||
total_query = select(func.count()).select_from(model).filter(*filter_conditions)
|
total_query = select(func.count()).select_from(model).filter(*filter_conditions)
|
||||||
total_result = await self.session.execute(total_query)
|
total_result = await self.session.execute(total_query)
|
||||||
total = total_result.scalar()
|
total = total_result.scalar()
|
||||||
|
|
||||||
query = query.limit(page_size).offset((page - 1) * page_size)
|
query = query.limit(page_size).offset((page - 1) * page_size)
|
||||||
result = await self.session.execute(query)
|
result = await self.session.execute(query)
|
||||||
records = result.scalars().all()
|
records = result.scalars().all()
|
||||||
|
|
||||||
return records, total
|
return records, total
|
||||||
|
|
||||||
@@ -52,103 +51,102 @@ class CRUD:
|
|||||||
self, group_by_field: str, filters: Dict[str, Any] = None, model: Model = None
|
self, group_by_field: str, filters: Dict[str, Any] = None, model: Model = None
|
||||||
) -> Dict[Any, int]:
|
) -> Dict[Any, int]:
|
||||||
model = model or self.model
|
model = model or self.model
|
||||||
filter_conditions = self.build_filter_conditions(filters) if filters else []
|
filter_conditions = self.build_filter_conditions(filters, model) if filters else []
|
||||||
group_by = getattr(model, group_by_field)
|
group_by = getattr(model, group_by_field)
|
||||||
query = select(group_by, func.count()).filter(*filter_conditions).group_by(group_by)
|
query = select(group_by, func.count()).filter(*filter_conditions).group_by(group_by)
|
||||||
|
|
||||||
async with self.session.begin():
|
result = await self.session.execute(query)
|
||||||
result = await self.session.execute(query)
|
records = result.all()
|
||||||
records = result.all()
|
|
||||||
|
|
||||||
return {record[0]: record[1] for record in records}
|
return {record[0]: record[1] for record in records}
|
||||||
|
|
||||||
async def create(self, data: dict, filters: dict, model: Model = None) -> Union[Model, bool]:
|
async def create(self, data: dict, filters: dict, model: Model = None) -> Union[Model, bool]:
|
||||||
model = model or self.model
|
model = model or self.model
|
||||||
async with self.session.begin():
|
|
||||||
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)
|
result = await self.session.execute(select(model).filter(
|
||||||
self.session.add(record)
|
*self.build_filter_conditions(filters, model))
|
||||||
await self.session.commit()
|
)
|
||||||
|
if result.scalars().first() is not None:
|
||||||
|
return False
|
||||||
|
|
||||||
|
record = model(**data)
|
||||||
|
self.session.add(record)
|
||||||
|
await self.session.commit()
|
||||||
return record
|
return record
|
||||||
|
|
||||||
async def get_all(self, filters: dict = None, model: Model = None) -> List[Model]:
|
async def get_all(self, filters: dict = None, model: Model = None) -> List[Model]:
|
||||||
model = model or self.model
|
model = model or self.model
|
||||||
async with self.session.begin():
|
|
||||||
result = await self.session.execute(
|
result = await self.session.execute(
|
||||||
select(model) if filters is None else select(model).filter(
|
select(model) if filters is None else select(model).filter(
|
||||||
*self.build_filter_conditions(filters, model)).order_by(desc(model.insert_time))
|
*self.build_filter_conditions(filters, model)).order_by(desc(model.insert_time))
|
||||||
)
|
)
|
||||||
records = result.scalars().all()
|
records = result.scalars().all()
|
||||||
return records
|
return records
|
||||||
|
|
||||||
async def update(self, data: dict, filters: dict, model: Model = None) -> Optional[Model]:
|
async def update(self, data: dict, filters: dict, model: Model = None) -> Optional[Model]:
|
||||||
model = model or self.model
|
model = model or self.model
|
||||||
async with self.session.begin():
|
|
||||||
result = await self.session.execute(select(model).filter(
|
|
||||||
*self.build_filter_conditions(filters, model))
|
|
||||||
)
|
|
||||||
record = result.scalars().first()
|
|
||||||
|
|
||||||
if not record:
|
result = await self.session.execute(select(model).filter(
|
||||||
return None
|
*self.build_filter_conditions(filters, model))
|
||||||
|
)
|
||||||
|
record = result.scalars().first()
|
||||||
|
|
||||||
count = 0
|
if not record:
|
||||||
for key, value in data.items():
|
return None
|
||||||
if any([
|
|
||||||
value is None,
|
count = 0
|
||||||
value == getattr(record, key),
|
for key, value in data.items():
|
||||||
key in ['created_at', 'updated_at']
|
if any([
|
||||||
]):
|
value is None,
|
||||||
continue
|
value == getattr(record, key),
|
||||||
setattr(record, key, value)
|
key in ['created_at', 'updated_at']
|
||||||
count += 1
|
]):
|
||||||
if count > 0:
|
continue
|
||||||
await self.session.commit()
|
setattr(record, key, value)
|
||||||
|
count += 1
|
||||||
|
if count > 0:
|
||||||
|
await self.session.commit()
|
||||||
return record
|
return record
|
||||||
|
|
||||||
async def updates(self, data: dict, filters: dict, model: Model = None) -> Optional[Model]:
|
async def updates(self, data: dict, filters: dict, model: Model = None) -> Optional[Model]:
|
||||||
model = model or self.model
|
model = model or self.model
|
||||||
async with self.session.begin():
|
|
||||||
result = await self.session.execute(select(model).filter(
|
|
||||||
*self.build_filter_conditions(filters, model))
|
|
||||||
)
|
|
||||||
records = result.scalars().all()
|
|
||||||
|
|
||||||
if len(records) == 0:
|
result = await self.session.execute(select(model).filter(
|
||||||
return None
|
*self.build_filter_conditions(filters, model))
|
||||||
count = 0
|
)
|
||||||
for record in records:
|
records = result.scalars().all()
|
||||||
for key, value in data.items():
|
|
||||||
if any([
|
if len(records) == 0:
|
||||||
value is None,
|
return None
|
||||||
value == getattr(record, key),
|
count = 0
|
||||||
key in ['insert_time', 'update_time']
|
for record in records:
|
||||||
]):
|
for key, value in data.items():
|
||||||
continue
|
if any([
|
||||||
setattr(record, key, value)
|
value is None,
|
||||||
count += 1
|
value == getattr(record, key),
|
||||||
if count > 0:
|
key in ['insert_time', 'update_time']
|
||||||
await self.session.commit()
|
]):
|
||||||
|
continue
|
||||||
|
setattr(record, key, value)
|
||||||
|
count += 1
|
||||||
|
if count > 0:
|
||||||
|
await self.session.commit()
|
||||||
return records
|
return records
|
||||||
|
|
||||||
async def delete(self, filters: dict, model: Model = None) -> bool:
|
async def delete(self, filters: dict, model: Model = None) -> bool:
|
||||||
model = model or self.model
|
model = model or self.model
|
||||||
async with self.session.begin():
|
|
||||||
result = await self.session.execute(select(model).filter(
|
|
||||||
*self.build_filter_conditions(filters, model))
|
|
||||||
)
|
|
||||||
record = result.scalars().first()
|
|
||||||
|
|
||||||
if not record:
|
result = await self.session.execute(select(model).filter(
|
||||||
return False
|
*self.build_filter_conditions(filters, model))
|
||||||
|
)
|
||||||
|
record = result.scalars().first()
|
||||||
|
|
||||||
await self.session.delete(record)
|
if not record:
|
||||||
await self.session.commit()
|
return False
|
||||||
|
|
||||||
|
await self.session.delete(record)
|
||||||
|
await self.session.commit()
|
||||||
return True
|
return True
|
||||||
|
|
||||||
def build_filter_conditions(self, filters: Dict[str, Any], model: Model) -> tuple:
|
def build_filter_conditions(self, filters: Dict[str, Any], model: Model) -> tuple:
|
||||||
|
|||||||
@@ -35,7 +35,8 @@ session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit
|
|||||||
|
|
||||||
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
|
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
|
||||||
async with session_maker() as session:
|
async with session_maker() as session:
|
||||||
yield session
|
async with session.begin():
|
||||||
|
yield session
|
||||||
|
|
||||||
|
|
||||||
def create_redis_pool():
|
def create_redis_pool():
|
||||||
|
|||||||
Reference in New Issue
Block a user