51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
from typing import Optional, List
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.audit_log import AuditLog
|
|
|
|
|
|
class AuditLogRepository:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def get_by_id(self, log_id: int) -> Optional[AuditLog]:
|
|
result = await self.db.execute(
|
|
select(AuditLog)
|
|
.options(selectinload(AuditLog.medicine), selectinload(AuditLog.user))
|
|
.where(AuditLog.id == log_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_list(
|
|
self,
|
|
medicine_id: Optional[int] = None,
|
|
user_id: Optional[int] = None,
|
|
action: Optional[str] = None,
|
|
page: int = 1,
|
|
page_size: int = 20
|
|
) -> List[AuditLog]:
|
|
query = select(AuditLog).options(
|
|
selectinload(AuditLog.medicine),
|
|
selectinload(AuditLog.user)
|
|
)
|
|
|
|
if medicine_id:
|
|
query = query.where(AuditLog.medicine_id == medicine_id)
|
|
if user_id:
|
|
query = query.where(AuditLog.user_id == user_id)
|
|
if action:
|
|
query = query.where(AuditLog.action == action)
|
|
|
|
query = query.order_by(AuditLog.created_at.desc())
|
|
query = query.offset((page - 1) * page_size).limit(page_size)
|
|
|
|
result = await self.db.execute(query)
|
|
return list(result.scalars().all())
|
|
|
|
async def create(self, data: dict) -> AuditLog:
|
|
log = AuditLog(**data)
|
|
self.db.add(log)
|
|
await self.db.flush()
|
|
return log |