84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
from datetime import datetime
|
||
from typing import Iterable, Optional
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy.sql.functions import func
|
||
from sqlalchemy.orm import joinedload
|
||
|
||
# from sqlalchemy.orm import selectinload
|
||
|
||
# from app.domain.models import AuditLog
|
||
from src.db.models.audit_log import AuditLog
|
||
|
||
|
||
class AuditLogRepository:
|
||
"""Репозиторий для работы с журналом аудита."""
|
||
|
||
def __init__(self, db: AsyncSession):
|
||
self.db = db
|
||
|
||
async def get(self, audit_log_id: int) -> Optional[AuditLog]:
|
||
"""Получение записи аудита по ID."""
|
||
return (
|
||
(
|
||
await self.db.execute(
|
||
select(AuditLog).where(AuditLog.id == audit_log_id).limit(1)
|
||
)
|
||
)
|
||
.scalars()
|
||
.first()
|
||
)
|
||
|
||
async def get_all(
|
||
self,
|
||
limit: int | None = None,
|
||
offset: int | None = None,
|
||
user_id: int | None = None,
|
||
org_unit_id: int | None = None,
|
||
task_id: int | None = None,
|
||
form_id: int | None = None,
|
||
event_type: str | None = None,
|
||
date_from: datetime | None = None,
|
||
date_to: datetime | None = None,
|
||
) -> tuple[Iterable[AuditLog], int]:
|
||
"""Получение всех записей аудита с опциональной фильтрацией."""
|
||
query = select(AuditLog)
|
||
query_count = select(func.count(AuditLog.id))
|
||
|
||
if user_id is not None:
|
||
query = query.where(AuditLog.user_id == user_id)
|
||
query_count = query_count.where(AuditLog.user_id == user_id)
|
||
if org_unit_id is not None:
|
||
query = query.where(AuditLog.org_unit_id == org_unit_id)
|
||
query_count = query_count.where(AuditLog.org_unit_id == org_unit_id)
|
||
if task_id is not None:
|
||
query = query.where(AuditLog.task_id == task_id)
|
||
query_count = query_count.where(AuditLog.task_id == task_id)
|
||
if form_id is not None:
|
||
query = query.where(AuditLog.form_id == form_id)
|
||
query_count = query_count.where(AuditLog.form_id == form_id)
|
||
if event_type is not None:
|
||
query = query.where(AuditLog.event_type == event_type)
|
||
query_count = query_count.where(AuditLog.event_type == event_type)
|
||
if date_from is not None:
|
||
query = query.where(AuditLog.event_dt >= date_from)
|
||
query_count = query_count.where(AuditLog.event_dt >= date_from)
|
||
if date_to is not None:
|
||
query = query.where(AuditLog.event_dt <= date_to)
|
||
query_count = query_count.where(AuditLog.event_dt <= date_to)
|
||
|
||
query = query.options(joinedload(AuditLog.user))
|
||
|
||
query = query.order_by(AuditLog.event_dt.desc())
|
||
|
||
if limit is not None:
|
||
query = query.limit(limit)
|
||
if offset is not None:
|
||
query = query.offset(offset)
|
||
|
||
return (await self.db.execute(query)).scalars().all(), (
|
||
await self.db.execute(query_count)
|
||
).scalar_one()
|
||
|