"""ReconConfig API proxy / fake endpoints. When RECON_API_BASE_URL is set in the environment, POST and PUT calls are forwarded to the real API. GET calls always use fake data in this workspace so the editor has something to load without a real database. Also provides a /api/configs/test-url endpoint that inspects file:// URLs and returns metadata (no file contents are returned). """ from __future__ import annotations import glob import os from datetime import datetime from pathlib import Path from typing import Any import httpx from fastapi import APIRouter, HTTPException, Request from fastapi.responses import JSONResponse from app.core.refdata import ReconConfigStatus, ReconPatterns from app.service.fake_configs import get_fake_configs router = APIRouter(prefix="/api/configs", tags=["ReconConfig"]) _RECON_API_BASE = os.getenv("RECON_API_BASE_URL", "").rstrip("/") # ── Helpers ──────────────────────────────────────────────────────────────── def _fake_config_full(ref: str, now: datetime) -> dict[str, Any] | None: """Build a full ReconConfigRequest-shaped dict from a FakeReconConfig.""" fake = next((c for c in get_fake_configs(now) if c.reference == ref), None) if fake is None: return None return { "reference": fake.reference, "name": fake.name, "business_process": fake.business_process, "comment": "", "data_type": "Unknown", "pattern": ReconPatterns.ONE_TO_ONE.value, "status": fake.status, "frequency": fake.frequency, "start_datetime": fake.start_datetime.isoformat() if fake.start_datetime else None, "field_mapping": {}, "sources": [ { "name": "Source System", "url": "file:///data/input/{{as_at_date.strftime('%Y%m%d')}}.csv", "comment": "", "day_offset": 0, "system_schema": {}, "obfuscate_fields": [], "pci_redact_fields": [], "field_widths": [], "profile_thresholds": [], "index_fields": [], "filter": "", "sql": "", "csv_spec": {"delimiter": ",", "header": True, "encoding": "utf_8", "trailer_rows": 0, "quoting": True}, "xml_spec": None, } ], "destination": { "name": "Destination System", "url": "mssql://GROUPDW?db=BIODS_PROCESSING&table=EXAMPLE", "comment": "", "day_offset": 0, "system_schema": {}, "obfuscate_fields": [], "pci_redact_fields": [], "field_widths": [], "profile_thresholds": [], "index_fields": [], "filter": "", "sql": "", "csv_spec": None, "xml_spec": None, }, } # ── List ──────────────────────────────────────────────────────────────────── @router.get("/") async def list_configs(): now = datetime.now() items = [] for c in get_fake_configs(now): items.append({ "reference": c.reference, "name": c.name, "status": c.status, "frequency": c.frequency, "business_process": c.business_process, }) return items # ── Get one ───────────────────────────────────────────────────────────────── @router.get("/{reference}") async def get_config(reference: str): now = datetime.now() if _RECON_API_BASE: async with httpx.AsyncClient(verify=False) as client: resp = await client.get(f"{_RECON_API_BASE}/api/configs/{reference}") if resp.status_code == 404: raise HTTPException(404, f"Config '{reference}' not found") return resp.json() data = _fake_config_full(reference, now) if data is None: raise HTTPException(404, f"Config '{reference}' not found") return data # ── Create ────────────────────────────────────────────────────────────────── @router.post("/") async def create_config(request: Request): body = await request.json() if _RECON_API_BASE: async with httpx.AsyncClient(verify=False) as client: resp = await client.post(f"{_RECON_API_BASE}/api/configs/", json=body) return JSONResponse(content=resp.json(), status_code=resp.status_code) # Workspace stub: echo back with a fake revision return JSONResponse( content={**body, "revision_number": 1, "user": {"username": "workspace"}}, status_code=201, ) # ── Update ────────────────────────────────────────────────────────────────── @router.put("/{reference}") async def update_config(reference: str, request: Request): body = await request.json() if _RECON_API_BASE: async with httpx.AsyncClient(verify=False) as client: resp = await client.put(f"{_RECON_API_BASE}/api/configs/{reference}", json=body) return JSONResponse(content=resp.json(), status_code=resp.status_code) return JSONResponse( content={**body, "revision_number": body.get("revision_number", 1) + 1, "user": {"username": "workspace"}}, status_code=200, ) # ── Test URL ──────────────────────────────────────────────────────────────── @router.post("/test-url") async def test_url(request: Request): """Return metadata for a URL pattern — no file contents ever returned.""" body = await request.json() url: str = body.get("url", "") as_at_date: str = body.get("as_at_date", datetime.now().strftime("%Y%m%d")) if not url.startswith("file://"): return {"type": "non-file", "message": "Only file:// URLs can be tested here. Database connections are validated at run time."} # Resolve the path part (strip file://) raw_path = url[7:] # Replace simple template variables for preview purposes from datetime import date as date_type try: as_at = datetime.strptime(as_at_date, "%Y%m%d").date() except ValueError: as_at = date_type.today() resolved = ( raw_path .replace("{{as_at_date.strftime('%Y%m%d')}}", as_at.strftime("%Y%m%d")) .replace("{{today.strftime('%Y%m%d')}}", date_type.today().strftime("%Y%m%d")) .replace("{{as_at_date.strftime('%Y-%m-%d')}}", as_at.strftime("%Y-%m-%d")) .replace("{{today.strftime('%Y-%m-%d')}}", date_type.today().strftime("%Y-%m-%d")) ) # Treat as glob pattern for anything still containing { if "*" in resolved or "?" in resolved or "{" in resolved: matches = glob.glob(resolved) if not matches: return {"type": "file", "resolved": resolved, "found": False, "message": "No files matched the pattern."} file_infos = [] for p in sorted(matches)[:10]: try: st = os.stat(p) file_infos.append({ "path": p, "size_bytes": st.st_size, "modified": datetime.fromtimestamp(st.st_mtime).isoformat(timespec="seconds"), }) except OSError: file_infos.append({"path": p, "error": "Could not stat file"}) return {"type": "file", "resolved": resolved, "found": True, "matches": len(matches), "files": file_infos} # Exact path p = Path(resolved) if not p.exists(): return {"type": "file", "resolved": resolved, "found": False, "message": f"File not found: {resolved}"} try: st = p.stat() return { "type": "file", "resolved": resolved, "found": True, "matches": 1, "files": [{ "path": str(p), "size_bytes": st.st_size, "modified": datetime.fromtimestamp(st.st_mtime).isoformat(timespec="seconds"), }], } except OSError as e: return {"type": "file", "resolved": resolved, "found": False, "message": str(e)}