19 lines
654 B
Python
19 lines
654 B
Python
"""Risk assessment persistence operations."""
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.entities import RiskAssessment
|
|
from app.repositories.base import BaseRepository
|
|
|
|
|
|
class RiskPredictionRepository(BaseRepository[RiskAssessment]):
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
super().__init__(session, RiskAssessment)
|
|
|
|
async def list_recent(self, *, limit: int = 20) -> list[RiskAssessment]:
|
|
statement = select(RiskAssessment).order_by(RiskAssessment.created_at.desc()).limit(limit)
|
|
result = await self.session.scalars(statement)
|
|
return list(result)
|
|
|