import csv import io import re import uuid from fastapi import HTTPException from sqlalchemy.ext.asyncio import AsyncSession from reports.models import ReportRuns, SavedReports from reports.runner import FILTER_KEYS, REPORT_TYPES, run_report from reports.serializers import serialize_report_run, serialize_saved_report def _as_uuid(value): if value in (None, ""): return None try: return uuid.UUID(str(value)) except (TypeError, ValueError): return None def _user_id(current_user): if not current_user or not current_user.get("id"): raise HTTPException(status_code=401, detail="Not authenticated") uid = _as_uuid(current_user["id"]) if uid is None: raise HTTPException(status_code=401, detail="Invalid user id") return uid def _clean_filters(filters): if filters is None: return {} if not isinstance(filters, dict): raise HTTPException(status_code=422, detail="filters must be an object") return {k: v for k, v in filters.items() if k in FILTER_KEYS and v not in (None, "")} def _validate_report_type(report_type): if report_type not in REPORT_TYPES: raise HTTPException( status_code=422, detail=f"report_type must be one of {', '.join(REPORT_TYPES)}" ) return report_type class Report: def __init__(self, session: AsyncSession): self.session = session async def get_reports(self, current_user): rows, total = await SavedReports.fetch_saved_reports( self.session, user_id=_user_id(current_user) ) return [serialize_saved_report(r) for r in rows], total async def create_report(self, payload, current_user): name = (payload.get("name") or "").strip() if not name: raise HTTPException(status_code=422, detail="name is required") report_type = _validate_report_type((payload.get("report_type") or "").strip()) row = await SavedReports.insert_saved_report(self.session, { "user_id": _user_id(current_user), "name": name, "description": (payload.get("description") or "").strip() or None, "report_type": report_type, "filters": _clean_filters(payload.get("filters")), }) return serialize_saved_report(row) async def update_report(self, record_id, payload, current_user): uid = _user_id(current_user) fields = {} if "name" in payload: name = (payload.get("name") or "").strip() if not name: raise HTTPException(status_code=422, detail="name cannot be blank") fields["name"] = name if "description" in payload: fields["description"] = (payload.get("description") or "").strip() or None if "report_type" in payload: fields["report_type"] = _validate_report_type((payload.get("report_type") or "").strip()) if "filters" in payload: fields["filters"] = _clean_filters(payload.get("filters")) if not fields: raise HTTPException(status_code=400, detail="No fields to update") row = await SavedReports.update_saved_report(self.session, record_id, fields, user_id=uid) if not row: raise HTTPException(status_code=404, detail="Saved report not found") return serialize_saved_report(row) async def delete_report(self, record_id, current_user): row = await SavedReports.soft_delete_saved_report( self.session, record_id, user_id=_user_id(current_user) ) if not row: raise HTTPException(status_code=404, detail="Saved report not found") return {"id": str(row.id), "deleted": True} async def _resolve_definition(self, record_id, report_type, filters, current_user): """A run targets either a saved report (by id) or an ad-hoc definition.""" if record_id: row = await SavedReports.get_by_id( self.session, record_id, user_id=_user_id(current_user) ) if not row: raise HTTPException(status_code=404, detail="Saved report not found") merged = dict(row.filters or {}) merged.update(_clean_filters(filters)) return row, row.report_type, merged, row.name report_type = _validate_report_type((report_type or "").strip()) return None, report_type, _clean_filters(filters), report_type async def run(self, current_user, *, record_id=None, report_type=None, filters=None, record_run=True, run_status="completed"): saved, resolved_type, resolved_filters, name = await self._resolve_definition( record_id, report_type, filters, current_user ) result = await run_report(self.session, resolved_type, resolved_filters) result["name"] = name if saved is not None: result["saved_report_id"] = str(saved.id) if record_run: await ReportRuns.insert_run(self.session, { "saved_report_id": saved.id, "run_by": _user_id(current_user), "params": resolved_filters, "row_count": result["row_count"], "status": run_status, }) await SavedReports.touch_last_run( self.session, saved.id, user_id=_user_id(current_user) ) return result async def export_csv(self, current_user, *, record_id=None, report_type=None, filters=None): result = await self.run( current_user, record_id=record_id, report_type=report_type, filters=filters, run_status="exported", ) buffer = io.StringIO() keys = [c["key"] for c in result["columns"]] writer = csv.writer(buffer, lineterminator="\r\n") writer.writerow([c["label"] for c in result["columns"]]) for row in result["rows"]: writer.writerow(["" if row.get(k) is None else row.get(k) for k in keys]) slug = re.sub(r"[^A-Za-z0-9_-]+", "-", result["name"]).strip("-").lower() or "report" return f"{slug}.csv", buffer.getvalue() async def get_runs(self, record_id, current_user): row = await SavedReports.get_by_id( self.session, record_id, user_id=_user_id(current_user) ) if not row: raise HTTPException(status_code=404, detail="Saved report not found") runs = await ReportRuns.fetch_runs(self.session, saved_report_id=row.id) return [serialize_report_run(r) for r in runs]