131 lines
4.3 KiB
Python
131 lines
4.3 KiB
Python
"""Persistence operations for system static data."""
|
|
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database import AsyncSessionLocal
|
|
from app.models.db import StaticData
|
|
|
|
|
|
class StaticDataService:
|
|
"""Create, read, update, and delete static data entries."""
|
|
|
|
@staticmethod
|
|
async def get_all_items(
|
|
session: Optional[AsyncSession] = None,
|
|
) -> List[StaticData]:
|
|
async def query(active_session: AsyncSession) -> List[StaticData]:
|
|
result = await active_session.execute(
|
|
select(StaticData).order_by(StaticData.category, StaticData.name)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
if session is not None:
|
|
return await query(session)
|
|
async with AsyncSessionLocal() as active_session:
|
|
return await query(active_session)
|
|
|
|
@staticmethod
|
|
async def create_static(
|
|
data: Dict[str, Any],
|
|
session: Optional[AsyncSession] = None,
|
|
) -> StaticData:
|
|
async def create(active_session: AsyncSession) -> StaticData:
|
|
item = StaticData(**data)
|
|
active_session.add(item)
|
|
await active_session.commit()
|
|
await active_session.refresh(item)
|
|
return item
|
|
|
|
if session is not None:
|
|
return await create(session)
|
|
async with AsyncSessionLocal() as active_session:
|
|
return await create(active_session)
|
|
|
|
@staticmethod
|
|
async def update_static(
|
|
item_id: int,
|
|
update_data: Dict[str, Any],
|
|
session: Optional[AsyncSession] = None,
|
|
) -> Optional[StaticData]:
|
|
async def update(active_session: AsyncSession) -> Optional[StaticData]:
|
|
result = await active_session.execute(
|
|
select(StaticData).where(StaticData.id == item_id)
|
|
)
|
|
item = result.scalar_one_or_none()
|
|
if item is None:
|
|
return None
|
|
|
|
for key, value in update_data.items():
|
|
if hasattr(item, key):
|
|
setattr(item, key, value)
|
|
|
|
await active_session.commit()
|
|
await active_session.refresh(item)
|
|
return item
|
|
|
|
if session is not None:
|
|
return await update(session)
|
|
async with AsyncSessionLocal() as active_session:
|
|
return await update(active_session)
|
|
|
|
@staticmethod
|
|
async def delete_static(
|
|
item_id: int,
|
|
session: Optional[AsyncSession] = None,
|
|
) -> bool:
|
|
async def delete(active_session: AsyncSession) -> bool:
|
|
result = await active_session.execute(
|
|
select(StaticData).where(StaticData.id == item_id)
|
|
)
|
|
item = result.scalar_one_or_none()
|
|
if item is None:
|
|
return False
|
|
|
|
await active_session.delete(item)
|
|
await active_session.commit()
|
|
return True
|
|
|
|
if session is not None:
|
|
return await delete(session)
|
|
async with AsyncSessionLocal() as active_session:
|
|
return await delete(active_session)
|
|
|
|
@staticmethod
|
|
async def get_by_category(
|
|
category: str,
|
|
session: Optional[AsyncSession] = None,
|
|
) -> List[StaticData]:
|
|
async def query(active_session: AsyncSession) -> List[StaticData]:
|
|
result = await active_session.execute(
|
|
select(StaticData)
|
|
.where(StaticData.category == category)
|
|
.order_by(StaticData.name)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
if session is not None:
|
|
return await query(session)
|
|
async with AsyncSessionLocal() as active_session:
|
|
return await query(active_session)
|
|
|
|
@staticmethod
|
|
async def get_all_categories(
|
|
session: Optional[AsyncSession] = None,
|
|
) -> List[str]:
|
|
async def query(active_session: AsyncSession) -> List[str]:
|
|
result = await active_session.execute(
|
|
select(StaticData.category).distinct()
|
|
)
|
|
return [row[0] for row in result]
|
|
|
|
if session is not None:
|
|
return await query(session)
|
|
async with AsyncSessionLocal() as active_session:
|
|
return await query(active_session)
|
|
|
|
|
|
static_data_service = StaticDataService()
|