from typing import Optional, List from sqlalchemy.ext.asyncio import AsyncSession from app.repositories.audit_log import AuditLogRepository class AuditService: def __init__(self, db: AsyncSession): self.db = db self.repo = AuditLogRepository(db) async def log_action( self, medicine_id: int, batch_id: Optional[int], user_id: Optional[int], action: str, quantity_change: int, quantity_after: int, remark: Optional[str] = None ): data = { "medicine_id": medicine_id, "batch_id": batch_id, "user_id": user_id, "action": action, "quantity_change": quantity_change, "quantity_after": quantity_after, "remark": remark } return await self.repo.create(data) async def get_audit_logs( self, medicine_id: Optional[int] = None, user_id: Optional[int] = None, action: Optional[str] = None, page: int = 1, page_size: int = 20 ) -> List: return await self.repo.get_list( medicine_id=medicine_id, user_id=user_id, action=action, page=page, page_size=page_size )