""" Repository for Diagnostic Report entities. """ from __future__ import annotations from collections.abc import Sequence from datetime import datetime from typing import cast from uuid import UUID from sqlalchemy import ColumnElement, and_, func, or_, select, text from sqlalchemy import case as sa_case from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.models.order import OrderPackage, OrderTest, OrderTestStatus, VisitOrder from app.models.package import TestPackage, TestPackageItem from app.models.report import Report, ReportStatus from app.models.test import CategoryModality, Test, TestCategory, TestParameter class ReportRepository: """Repository handling database operations for Report.""" @staticmethod def _has_abnormal_flag() -> ColumnElement[bool]: """EXISTS check for ≥1 analyte with is_abnormal=true in result_data.parameters.""" return cast( ColumnElement[bool], text( "EXISTS (SELECT 1 FROM jsonb_each(report.result_data -> 'parameters') AS p(k, v) " "WHERE (v ->> 'is_abnormal')::boolean IS TRUE)" ), ) async def get_by_id(self, db: AsyncSession, report_id: UUID) -> Report | None: """Fetch report with order, patient, and verifier loaded.""" stmt = ( select(Report) .where(Report.id == report_id, Report.deleted_at.is_(None)) .options( selectinload(Report.verifier), selectinload(Report.order).selectinload(VisitOrder.patient), selectinload(Report.order).selectinload(VisitOrder.lab), selectinload(Report.order).selectinload(VisitOrder.samples), selectinload(Report.order) .selectinload(VisitOrder.order_tests) .selectinload(OrderTest.test) .selectinload(Test.parameters) .selectinload(TestParameter.ranges), selectinload(Report.order) .selectinload(VisitOrder.order_tests) .selectinload(OrderTest.test) .selectinload(Test.category), selectinload(Report.order) .selectinload(VisitOrder.order_packages) .selectinload(OrderPackage.package), selectinload(Report.order) .selectinload(VisitOrder.order_packages) .selectinload(OrderPackage.package) .selectinload(TestPackage.items), selectinload(Report.order) .selectinload(VisitOrder.order_packages) .selectinload(OrderPackage.package) .selectinload(TestPackage.items) .selectinload(TestPackageItem.test) .selectinload(Test.parameters) .selectinload(TestParameter.ranges), selectinload(Report.order) .selectinload(VisitOrder.order_packages) .selectinload(OrderPackage.package) .selectinload(TestPackage.items) .selectinload(TestPackageItem.test) .selectinload(Test.category), ) ) res = await db.execute(stmt) return res.scalar_one_or_none() async def get_latest_by_order_id( self, db: AsyncSession, order_id: UUID, modality: str | None = None ) -> Report | None: """Fetch the most recent version of a report for an order (optionally one side).""" stmt = ( select(Report) .where(Report.order_id == order_id, Report.deleted_at.is_(None)) .options( selectinload(Report.verifier), selectinload(Report.order).selectinload(VisitOrder.patient), selectinload(Report.order).selectinload(VisitOrder.order_tests).selectinload(OrderTest.test), ) .order_by(Report.version.desc()) .limit(1) ) if modality in ("imaging", "pathology"): stmt = stmt.where(Report.modality == CategoryModality(modality)) res = await db.execute(stmt) return res.scalar_one_or_none() async def get_by_order_and_version( self, db: AsyncSession, order_id: UUID, version: int, ) -> Report | None: """Fetch specific version of report for an order.""" stmt = ( select(Report) .where( Report.order_id == order_id, Report.version == version, Report.deleted_at.is_(None), ) .options( selectinload(Report.verifier), selectinload(Report.order).selectinload(VisitOrder.patient), ) ) res = await db.execute(stmt) return res.scalar_one_or_none() async def get_all_versions_by_order_id( self, db: AsyncSession, order_id: UUID, modality: str | None = None, ) -> list[Report]: """Fetch all report versions for an order (optionally one side), version descending.""" stmt = ( select(Report) .where(Report.order_id == order_id, Report.deleted_at.is_(None)) .options( selectinload(Report.verifier), selectinload(Report.order).selectinload(VisitOrder.patient), ) .order_by(Report.version.desc()) ) if modality in ("imaging", "pathology"): stmt = stmt.where(Report.modality == CategoryModality(modality)) res = await db.execute(stmt) return list(res.scalars().all()) async def modality_by_order_ids(self, db: AsyncSession, order_ids: list[UUID]) -> dict[UUID, str]: """Classify orders by discipline mix: pathology, imaging, or mixed. Mixed (and empty) orders resolve to "mixed" so both queues stay safe. """ if not order_ids: return {} stmt = ( select( OrderTest.order_id, func.coalesce( func.sum( sa_case( (TestCategory.modality == CategoryModality.IMAGING, 1), else_=0, ) ), 0, ).label("imaging_n"), func.count(OrderTest.id).label("total_n"), ) .join(Test, OrderTest.test_id == Test.id) .join(TestCategory, Test.category_id == TestCategory.id) .where( OrderTest.order_id.in_(order_ids), OrderTest.status != OrderTestStatus.CANCELLED, ) .group_by(OrderTest.order_id) ) rows = (await db.execute(stmt)).all() result: dict[UUID, str] = {} for order_id, imaging_n, total_n in rows: imaging_n = int(imaging_n or 0) total_n = int(total_n or 0) if imaging_n > 0 and imaging_n == total_n: result[order_id] = "imaging" elif imaging_n == 0 and total_n > 0: result[order_id] = "pathology" else: result[order_id] = "mixed" return result async def list( self, db: AsyncSession, lab_id: UUID | None = None, status: ReportStatus | None = None, statuses: Sequence[ReportStatus] | None = None, order_id: UUID | None = None, patient_id: UUID | None = None, activity_from: datetime | None = None, activity_to: datetime | None = None, critical_only: bool = False, modality: str | None = None, offset: int = 0, limit: int = 20, ) -> list[Report]: """List reports with filters and pagination.""" stmt = ( select(Report) .join(VisitOrder, Report.order_id == VisitOrder.id) .where(Report.deleted_at.is_(None)) .options( selectinload(Report.verifier), selectinload(Report.order).selectinload(VisitOrder.patient), ) .order_by(Report.created_at.desc()) ) conditions: list[ColumnElement[bool]] = [] if lab_id: conditions.append(VisitOrder.lab_id == lab_id) if statuses: conditions.append(Report.status.in_(statuses)) elif status: conditions.append(Report.status == status) if status == ReportStatus.REJECTED or (statuses and ReportStatus.REJECTED in statuses): conditions.append(~Report.amendments.any(Report.deleted_at.is_(None))) if order_id: conditions.append(Report.order_id == order_id) if patient_id: conditions.append(VisitOrder.patient_id == patient_id) if activity_from is not None or activity_to is not None: activity_conditions: list[ColumnElement[bool]] = [] for field in (Report.created_at, Report.verified_at, Report.released_at): bounds: list[ColumnElement[bool]] = [] if activity_from is not None: bounds.append(field >= activity_from) if activity_to is not None: bounds.append(field < activity_to) activity_conditions.append(and_(*bounds)) conditions.append(or_(*activity_conditions)) if critical_only: conditions.append(self._has_abnormal_flag()) if modality in ("imaging", "pathology"): conditions.append(Report.modality == CategoryModality(modality)) if conditions: stmt = stmt.where(*conditions) stmt = stmt.offset(offset).limit(limit) res = await db.execute(stmt) return list(res.scalars().all()) async def count( self, db: AsyncSession, lab_id: UUID | None = None, status: ReportStatus | None = None, statuses: Sequence[ReportStatus] | None = None, order_id: UUID | None = None, patient_id: UUID | None = None, activity_from: datetime | None = None, activity_to: datetime | None = None, critical_only: bool = False, modality: str | None = None, ) -> int: """Count total reports matching filters.""" stmt = ( select(func.count()) .select_from(Report) .join(VisitOrder, Report.order_id == VisitOrder.id) .where(Report.deleted_at.is_(None)) ) conditions: list[ColumnElement[bool]] = [] if lab_id: conditions.append(VisitOrder.lab_id == lab_id) if statuses: conditions.append(Report.status.in_(statuses)) elif status: conditions.append(Report.status == status) if status == ReportStatus.REJECTED or (statuses and ReportStatus.REJECTED in statuses): conditions.append(~Report.amendments.any(Report.deleted_at.is_(None))) if order_id: conditions.append(Report.order_id == order_id) if patient_id: conditions.append(VisitOrder.patient_id == patient_id) if activity_from is not None or activity_to is not None: activity_conditions: list[ColumnElement[bool]] = [] for field in (Report.created_at, Report.verified_at, Report.released_at): bounds: list[ColumnElement[bool]] = [] if activity_from is not None: bounds.append(field >= activity_from) if activity_to is not None: bounds.append(field < activity_to) activity_conditions.append(and_(*bounds)) conditions.append(or_(*activity_conditions)) if critical_only: conditions.append(self._has_abnormal_flag()) if modality in ("imaging", "pathology"): conditions.append(Report.modality == CategoryModality(modality)) if conditions: stmt = stmt.where(*conditions) res = await db.execute(stmt) return int(res.scalar_one() or 0) async def modalities_present(self, db: AsyncSession, order_id: UUID) -> set[str]: """Discipline sides already owning at least one report for an order.""" stmt = select(Report.modality).where(Report.order_id == order_id, Report.deleted_at.is_(None)) rows = (await db.execute(stmt)).all() out: set[str] = set() for (modality,) in rows: out.add(modality.value if hasattr(modality, "value") else str(modality)) return out async def create(self, db: AsyncSession, report: Report) -> Report: """Persist a new report record.""" db.add(report) await db.flush() await db.refresh(report) return report