50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
"""时态纪念日数据访问（辰正 §5.3）。"""
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.time_anniversary import TimeAnniversary
|
|
|
|
|
|
async def create_anniversary(
|
|
session: AsyncSession, anniversary: TimeAnniversary
|
|
) -> TimeAnniversary:
|
|
session.add(anniversary)
|
|
await session.flush()
|
|
await session.refresh(anniversary)
|
|
return anniversary
|
|
|
|
|
|
async def get_anniversary(session: AsyncSession, anniversary_id: str) -> TimeAnniversary | None:
|
|
return await session.get(TimeAnniversary, anniversary_id)
|
|
|
|
|
|
async def list_anniversaries(session: AsyncSession, character_id: str) -> list[TimeAnniversary]:
|
|
result = await session.execute(
|
|
select(TimeAnniversary)
|
|
.where(TimeAnniversary.character_id == character_id)
|
|
.order_by(TimeAnniversary.date.asc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def update_anniversary(
|
|
session: AsyncSession, anniversary_id: str, **fields: object
|
|
) -> TimeAnniversary | None:
|
|
anniversary = await session.get(TimeAnniversary, anniversary_id)
|
|
if anniversary is None:
|
|
return None
|
|
for key, value in fields.items():
|
|
setattr(anniversary, key, value)
|
|
await session.flush()
|
|
return anniversary
|
|
|
|
|
|
async def delete_anniversary(session: AsyncSession, anniversary_id: str) -> bool:
|
|
anniversary = await session.get(TimeAnniversary, anniversary_id)
|
|
if anniversary is None:
|
|
return False
|
|
await session.delete(anniversary)
|
|
await session.flush()
|
|
return True
|