fhir_upload.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = [
#   "requests>=2.28.0",
#   "pyyaml>=6.0",
#   "click>=8.0",
#   "tenacity>=8.0",
#   "tabulate>=0.9.0",
# ]
# ///
"""FHIR Resource Uploader

Uploads FHIR resources from a local directory to a FHIR server
via the FHIR Gateway with Keycloak authentication.

Usage:
    python fhir_upload.py [OPTIONS]

See --help for full usage, or README.md for operator guide.
"""

import json
import mimetypes
import os
import re
import subprocess
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

import click
import requests
import yaml
from tabulate import tabulate
from tenacity import Retrying, retry_if_exception_type, stop_after_attempt

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

TRANSIENT_STATUS_CODES = {429, 500, 502, 503, 504}
REQUEST_TIMEOUT = 30   # seconds per HTTP request
MAX_ATTEMPTS = 4       # 1 initial + 3 retries (delays: 1s, 2s, 4s)

MIME_TYPES = {
    ".json": "application/json",
    ".html": "text/html",
    ".xml": "application/xml",
    ".txt": "text/plain",
    ".properties": "text/plain",
}

JUNK_EXTENSIONS = {".swp", ".tmp", ".bak"}

ENV_OVERRIDES = {
    "FHIR_UPLOAD_USERNAME": "username",
    "FHIR_UPLOAD_PASSWORD": "password",
    "FHIR_UPLOAD_CLIENT_SECRET": "client_secret",
}


# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------

@dataclass
class FolderEntry:
    name: str
    resource_type: str | None = None


@dataclass
class AppConfig:
    fhir_base_url: str
    token_url: str
    client_id: str
    grant_type: str
    username: str = ""
    password: str = ""
    client_secret: str = ""
    resource_dir: str = "."
    processing_order: list[FolderEntry] = field(default_factory=list)


@dataclass
class ResourceFile:
    path: Path
    folder: str
    resource_type: str
    resource_id: str
    raw: bytes
    content_type: str = "application/fhir+json"
    rel_path: str = ""  # path relative to resource_dir, set by discover_resources

    @property
    def key(self) -> str:
        return f"{self.resource_type}/{self.resource_id}"


@dataclass
class UploadResult:
    folder: str
    resource: str
    status: str  # "ok" | "failed" | "skipped" | "invalid" | "dry-run"
    code: Optional[int] = None
    reason: str = ""
    response_body: str = ""
    file_path: str = ""  # path relative to resource_dir (empty for invalid files)


# ---------------------------------------------------------------------------
# Config loading
# ---------------------------------------------------------------------------

def load_config(config_path: str) -> AppConfig:
    path = Path(config_path)
    if not path.exists():
        raise click.ClickException(f"Config file not found: {config_path}")

    with open(path, encoding="utf-8") as f:
        raw = yaml.safe_load(f)

    if not isinstance(raw, dict):
        raise click.ClickException("Config file must be a YAML mapping")

    kc = raw.get("keycloak") or {}
    grant_type = kc.get("grant_type", "password")

    processing_order = _parse_processing_order(raw.get("processing_order") or [])

    cfg = AppConfig(
        fhir_base_url=raw.get("fhir_base_url") or "",
        token_url=kc.get("token_url") or "",
        client_id=kc.get("client_id") or "",
        grant_type=grant_type,
        username=kc.get("username") or "",
        password=kc.get("password") or "",
        client_secret=kc.get("client_secret") or "",
        resource_dir=str(
            (Path(config_path).resolve().parent / (raw.get("resource_dir") or ".")).resolve()
        ),
        processing_order=processing_order,
    )

    # Environment variable overrides take precedence over YAML values
    for env_var, attr in ENV_OVERRIDES.items():
        val = os.environ.get(env_var)
        if val is not None:
            setattr(cfg, attr, val)

    _validate_config(cfg)
    _warn_insecure_urls(cfg)
    return cfg


def _parse_processing_order(entries: list) -> list[FolderEntry]:
    result: list[FolderEntry] = []
    for i, entry in enumerate(entries):
        if entry is None:
            click.echo(f"WARNING: Ignoring empty entry at processing_order[{i}]", err=True)
            continue
        if isinstance(entry, str):
            result.append(FolderEntry(name=entry))
        elif isinstance(entry, dict):
            unknown = set(entry.keys()) - {"name", "resource_type"}
            if unknown:
                raise click.ClickException(
                    f"processing_order[{i}]: unknown keys: {', '.join(sorted(unknown))}"
                )
            name = entry.get("name")
            if not name or not isinstance(name, str):
                raise click.ClickException(
                    f"processing_order[{i}]: 'name' must be a non-empty string"
                )
            resource_type = entry.get("resource_type")
            if resource_type is not None and (
                not isinstance(resource_type, str) or not resource_type.strip()
            ):
                raise click.ClickException(
                    f"processing_order[{i}]: 'resource_type' must be a non-empty string"
                )
            result.append(FolderEntry(name=name, resource_type=resource_type))
        else:
            raise click.ClickException(
                f"processing_order[{i}]: entry must be a string or mapping, got {type(entry).__name__}"
            )
    return result


def slugify_id(filename: str) -> str:
    stem = Path(filename).stem
    s = stem.lower()
    s = re.sub(r"[^a-z0-9.\-]", "-", s)
    s = re.sub(r"-{2,}", "-", s)
    s = s.strip("-.")
    return s[:64]


def _get_mime_type(file_path: Path) -> str:
    ext = file_path.suffix.lower()
    if ext in MIME_TYPES:
        return MIME_TYPES[ext]
    guessed, _ = mimetypes.guess_type(file_path.name)
    return guessed or "application/octet-stream"


def _validate_config(cfg: AppConfig) -> None:
    errors = []

    if not cfg.fhir_base_url:
        errors.append("fhir_base_url is required")
    if not cfg.token_url:
        errors.append("keycloak.token_url is required")
    if not cfg.client_id:
        errors.append("keycloak.client_id is required")

    if cfg.grant_type == "password":
        if not cfg.username:
            errors.append("keycloak.username is required (or set FHIR_UPLOAD_USERNAME)")
        if not cfg.password:
            errors.append("keycloak.password is required (or set FHIR_UPLOAD_PASSWORD)")
    elif cfg.grant_type == "client_credentials":
        if not cfg.client_secret:
            errors.append("keycloak.client_secret is required (or set FHIR_UPLOAD_CLIENT_SECRET)")
    else:
        errors.append(
            f"keycloak.grant_type must be 'password' or 'client_credentials', got: {cfg.grant_type!r}"
        )

    if not cfg.processing_order:
        errors.append("processing_order must be a non-empty list")

    if not Path(cfg.resource_dir).exists():
        errors.append(f"resource_dir does not exist: {cfg.resource_dir!r}")

    if errors:
        raise click.ClickException("Config validation failed:\n  " + "\n  ".join(errors))


def _warn_insecure_urls(cfg: AppConfig) -> None:
    localhost = ("http://localhost", "http://127.0.0.1", "http://[::1]")
    for label, url in [
        ("fhir_base_url", cfg.fhir_base_url),
        ("keycloak.token_url", cfg.token_url),
    ]:
        if url.startswith("http://") and not any(url.startswith(p) for p in localhost):
            click.echo(
                f"WARNING: {label} uses HTTP on a non-localhost address.\n"
                "  Credentials and tokens will be transmitted in the clear.",
                err=True,
            )


# ---------------------------------------------------------------------------
# Authentication
# ---------------------------------------------------------------------------

class AuthClient:
    def __init__(self, cfg: AppConfig):
        self._cfg = cfg
        self.session = requests.Session()
        self._refresh_token: Optional[str] = None

    def _set_token(self, access_token: str, refresh_token: Optional[str]) -> None:
        self.session.headers["Authorization"] = f"Bearer {access_token}"
        self._refresh_token = refresh_token

    def authenticate(self) -> None:
        """Obtain a fresh access token from Keycloak."""
        tokens = self._post_token(self._build_grant_data())
        self._set_token(tokens["access_token"], tokens.get("refresh_token"))

    def refresh_or_reauthenticate(self) -> None:
        """Called on 401 -- try refresh token first, then fall back to full re-auth."""
        if self._refresh_token and self._try_refresh():
            return
        click.echo("  Re-authenticating...", err=True)
        self.authenticate()

    def _try_refresh(self) -> bool:
        data = {
            "grant_type": "refresh_token",
            "client_id": self._cfg.client_id,
            "refresh_token": self._refresh_token,
        }
        if self._cfg.client_secret:
            data["client_secret"] = self._cfg.client_secret
        try:
            tokens = self._post_token(data)
            self._set_token(tokens["access_token"], tokens.get("refresh_token", self._refresh_token))
            return True
        except click.ClickException:
            return False

    def _build_grant_data(self) -> dict:
        data: dict = {
            "grant_type": self._cfg.grant_type,
            "client_id": self._cfg.client_id,
        }
        if self._cfg.grant_type == "password":
            data["username"] = self._cfg.username
            data["password"] = self._cfg.password
        if self._cfg.client_secret:
            data["client_secret"] = self._cfg.client_secret
        return data

    def _post_token(self, data: dict) -> dict:
        try:
            resp = requests.post(self._cfg.token_url, data=data, timeout=REQUEST_TIMEOUT)
        except requests.exceptions.ConnectionError:
            raise click.ClickException(f"Cannot connect to Keycloak at {self._cfg.token_url}")
        except requests.exceptions.Timeout:
            raise click.ClickException("Timeout connecting to Keycloak")

        try:
            body = resp.json()
        except ValueError:
            raise click.ClickException(
                f"Keycloak returned non-JSON response (HTTP {resp.status_code})"
            )

        if "error" in body:
            raise click.ClickException(
                f"Authentication failed: {body.get('error_description') or body['error']}"
            )
        if resp.status_code != 200:
            raise click.ClickException(f"Authentication failed with HTTP {resp.status_code}")

        return body


# ---------------------------------------------------------------------------
# Git pull
# ---------------------------------------------------------------------------

def git_pull(resource_dir: str, no_pull: bool) -> None:
    if no_pull:
        return

    if not (Path(resource_dir) / ".git").exists():
        click.echo(f"INFO: {resource_dir!r} is not a Git repository -- skipping git pull.")
        return

    try:
        result = subprocess.run(
            ["git", "pull"], cwd=resource_dir, capture_output=True, text=True
        )
    except FileNotFoundError:
        raise click.ClickException("git is not installed or not found in PATH")

    if result.returncode != 0:
        click.echo(result.stderr.strip() or result.stdout.strip(), err=True)
        raise click.ClickException("git pull failed -- resolve the issue and re-run")

    click.echo(f"git pull: {result.stdout.strip() or 'Already up to date.'}")


# ---------------------------------------------------------------------------
# Resource discovery & validation
# ---------------------------------------------------------------------------

def _is_junk_file(file_path: Path) -> bool:
    if file_path.suffix.lower() in JUNK_EXTENSIONS:
        return True
    return any(part.startswith(".") for part in file_path.parts)


def discover_resources(
    cfg: AppConfig, folder_filter: Optional[str]
) -> tuple[list[ResourceFile], list[UploadResult]]:
    folders = cfg.processing_order
    subpath_filter: str | None = None

    if folder_filter:
        # Split by first "/" to extract optional subpath
        parts = folder_filter.split("/", 1)
        filter_name = parts[0]
        subpath_filter = parts[1] if len(parts) > 1 else None

        matched = [e for e in folders if e.name == filter_name]
        if not matched:
            available = ", ".join(e.name for e in folders)
            raise click.ClickException(
                f"Folder {filter_name!r} is not in processing_order. "
                f"Available: {available}"
            )
        folders = matched

    resources: list[ResourceFile] = []
    issues: list[UploadResult] = []
    seen: dict[str, str] = {}  # key -> path relative to resource_dir

    for entry in folders:
        folder_name = entry.name
        folder_path = Path(cfg.resource_dir) / folder_name

        if not folder_path.exists():
            click.echo(f"WARNING: Folder not found: {folder_path}", err=True)
            continue

        # Determine which files to scan
        if subpath_filter:
            target = folder_path / subpath_filter
            if target.is_file():
                files = [target]
            elif target.is_dir():
                if entry.resource_type:
                    files = sorted(f for f in target.rglob("*.*") if not _is_junk_file(f))
                else:
                    files = sorted(target.rglob("*.json"))
            else:
                raise click.ClickException(
                    f"Subpath '{subpath_filter}' not found in folder '{folder_name}'"
                )
        elif entry.resource_type:
            files = sorted(f for f in folder_path.rglob("*.*") if not _is_junk_file(f))
        else:
            files = sorted(folder_path.rglob("*.json"))

        if not files:
            continue

        for file_path in files:
            parsed = _parse_resource_file(
                file_path, folder_name, folder_path,
                resource_type_override=entry.resource_type,
            )

            if isinstance(parsed, UploadResult):
                issues.append(parsed)
                continue

            rel = str(file_path.relative_to(Path(cfg.resource_dir)))
            if parsed.key in seen:
                issues.append(UploadResult(
                    folder=folder_name,
                    resource=parsed.key,
                    status="skipped",
                    file_path=rel,
                    reason=f"Duplicate (first seen at {seen[parsed.key]})",
                ))
                continue

            seen[parsed.key] = rel
            parsed.rel_path = rel
            resources.append(parsed)

    return resources, issues


def _parse_resource_file(
    file_path: Path, folder_name: str, folder_path: Path,
    resource_type_override: str | None = None,
) -> ResourceFile | UploadResult:
    """Parse and validate a resource file. Returns ResourceFile on success."""
    display_name = str(file_path.relative_to(folder_path))

    def fail(reason: str) -> UploadResult:
        return UploadResult(
            folder=folder_name, resource=display_name, status="invalid", reason=reason
        )

    try:
        raw = file_path.read_bytes()
    except OSError as e:
        return fail(str(e))

    if resource_type_override:
        resource_id = slugify_id(file_path.name)
        if not resource_id:
            return fail("Cannot derive a valid resource ID from filename")
        return ResourceFile(
            path=file_path,
            folder=folder_name,
            resource_type=resource_type_override,
            resource_id=resource_id,
            raw=raw,
            content_type=_get_mime_type(file_path),
        )

    if raw.startswith(b"\xef\xbb\xbf"):  # Strip UTF-8 BOM
        raw = raw[3:]

    try:
        data = json.loads(raw.decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError) as e:
        return fail(f"Invalid JSON: {e}")

    if not isinstance(data, dict):
        return fail("JSON root must be an object")

    resource_type = data.get("resourceType")
    if not resource_type or not isinstance(resource_type, str):
        return fail("Missing or empty 'resourceType' field")

    resource_id = data.get("id")
    if not resource_id or not isinstance(resource_id, str):
        return fail("Missing or empty 'id' field")

    return ResourceFile(
        path=file_path,
        folder=folder_name,
        resource_type=resource_type,
        resource_id=resource_id,
        raw=raw,
    )


# ---------------------------------------------------------------------------
# Upload
# ---------------------------------------------------------------------------

class _RetryableError(Exception):
    """Raised for transient errors that tenacity should retry."""

    def __init__(self, exc_or_response, retry_after: Optional[float] = None):
        self.original = exc_or_response
        self.retry_after = retry_after
        self.response: Optional[requests.Response] = (
            exc_or_response if isinstance(exc_or_response, requests.Response) else None
        )


def _tenacity_wait(retry_state) -> float:
    """Exponential backoff (1s, 2s, 4s), overridden by Retry-After header for 429."""
    exc = retry_state.outcome.exception()
    if isinstance(exc, _RetryableError) and exc.retry_after:
        return exc.retry_after
    # attempt_number is 1-indexed; gives waits of 1, 2, 4 seconds
    return 2 ** (retry_state.attempt_number - 1)


def upload_resource(
    resource: ResourceFile, auth: AuthClient, base_url: str, verbose: bool
) -> UploadResult:
    url = f"{base_url.rstrip('/')}/{resource.resource_type}/{resource.resource_id}"

    def put_once() -> requests.Response:
        """Single PUT attempt; handles one 401 re-auth inline (not a tenacity retry)."""
        try:
            resp = auth.session.put(
                url, data=resource.raw,
                headers={"Content-Type": resource.content_type},
                timeout=REQUEST_TIMEOUT,
            )
        except requests.exceptions.RequestException as e:
            raise _RetryableError(e)

        if resp.status_code == 401:
            auth.refresh_or_reauthenticate()
            try:
                resp = auth.session.put(
                    url, data=resource.raw,
                    headers={"Content-Type": resource.content_type},
                    timeout=REQUEST_TIMEOUT,
                )
            except requests.exceptions.RequestException as e:
                raise _RetryableError(e)

        if resp.status_code in TRANSIENT_STATUS_CODES:
            retry_after: Optional[float] = None
            try:
                retry_after = float(resp.headers.get("Retry-After", 0)) or None
            except (ValueError, TypeError):
                pass
            raise _RetryableError(resp, retry_after=retry_after)

        return resp

    try:
        for attempt in Retrying(
            retry=retry_if_exception_type(_RetryableError),
            stop=stop_after_attempt(MAX_ATTEMPTS),
            wait=_tenacity_wait,
            reraise=True,
        ):
            with attempt:
                resp = put_once()
    except _RetryableError as e:
        if e.response is not None:
            return UploadResult(
                folder=resource.folder,
                resource=resource.key,
                status="failed",
                code=e.response.status_code,
                reason=e.response.reason,
                response_body=e.response.text if verbose else "",
                file_path=resource.rel_path,
            )
        return UploadResult(
            folder=resource.folder,
            resource=resource.key,
            status="failed",
            reason=str(e.original),
            file_path=resource.rel_path,
        )

    if resp.status_code >= 400:
        return UploadResult(
            folder=resource.folder,
            resource=resource.key,
            status="failed",
            code=resp.status_code,
            reason=resp.reason,
            response_body=resp.text if verbose else "",
            file_path=resource.rel_path,
        )

    return UploadResult(
        folder=resource.folder,
        resource=resource.key,
        status="ok",
        code=resp.status_code,
        reason=resp.reason,
        file_path=resource.rel_path,
    )


# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------

def print_summary(
    results: list[UploadResult], issues: list[UploadResult], verbose: bool
) -> int:
    all_results = results + issues

    # Group by folder, preserving processing order
    by_folder: dict[str, list[UploadResult]] = {}
    for r in all_results:
        by_folder.setdefault(r.folder, []).append(r)

    click.echo("\nUpload Summary")
    click.echo("-" * 60)

    for folder_name, folder_results in by_folder.items():
        click.echo(f"Folder: {folder_name}")
        rows = []
        for r in folder_results:
            if r.status == "ok":
                icon, action = "[OK]", "PUT"
                detail = f"{r.code} {r.reason}"
            elif r.status == "dry-run":
                icon, action = "[--]", "PUT"
                detail = "Would upload"
            elif r.status == "skipped":
                icon, action = "[!!]", "SKIP"
                detail = r.reason
            else:  # "failed" or "invalid"
                icon = "[FAIL]"
                action = "PUT" if r.status == "failed" else "SKIP"
                detail = f"{r.code} {r.reason}" if r.code else r.reason
            rows.append([f"  {icon}", action, r.resource, detail])
            if r.file_path:
                rows.append(["", "", f"    {r.file_path}", ""])
        click.echo(tabulate(rows, tablefmt="plain"))

        if verbose:
            for r in folder_results:
                if r.response_body:
                    click.echo(f"\n  Response body for {r.resource}:\n  {r.response_body[:1000]}")

    click.echo("-" * 60)

    n_ok = sum(1 for r in results if r.status == "ok")
    n_dry = sum(1 for r in results if r.status == "dry-run")
    n_failed = sum(1 for r in results if r.status == "failed")
    n_skipped = sum(1 for r in all_results if r.status == "skipped")
    n_invalid = sum(1 for r in all_results if r.status == "invalid")
    n_total = len(all_results)

    parts = [f"Total: {n_total} files"]
    if n_dry:
        parts.append(f"{n_dry} would upload (dry-run)")
    else:
        parts.append(f"{n_ok} uploaded")
    if n_failed:
        parts.append(f"{n_failed} failed")
    if n_skipped:
        parts.append(f"{n_skipped} duplicate")
    if n_invalid:
        parts.append(f"{n_invalid} invalid")
    click.echo(" | ".join(parts))

    return 1 if n_failed > 0 or n_invalid > 0 else 0


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

@click.command()
@click.option(
    "--config", "config_path",
    default="upload-config.yaml", show_default=True,
    help="Path to config YAML file.",
)
@click.option(
    "--dry-run", is_flag=True,
    help="Authenticate, validate, and show what would be uploaded -- without uploading.",
)
@click.option(
    "--no-pull", is_flag=True,
    help="Skip git pull before uploading.",
)
@click.option(
    "--folder", "folder_filter",
    default=None, metavar="FOLDER[/PATH]",
    help="Upload only this folder or subpath (e.g. 'app', 'app/profiles', 'app/file.json'). Folder must be listed in processing_order.",
)
@click.option(
    "--verbose", is_flag=True,
    help="Show full response bodies on failure.",
)
def main(
    config_path: str,
    dry_run: bool,
    no_pull: bool,
    folder_filter: Optional[str],
    verbose: bool,
) -> None:
    """Upload FHIR resources to a FHIR server via Keycloak authentication.

    Reads configuration from upload-config.yaml (override with --config).
    Sensitive fields (password, client_secret, username) can be supplied via
    FHIR_UPLOAD_PASSWORD, FHIR_UPLOAD_CLIENT_SECRET, FHIR_UPLOAD_USERNAME
    environment variables instead of the config file.
    """
    cfg = load_config(config_path)

    git_pull(cfg.resource_dir, no_pull)

    click.echo("Scanning resources...")
    resources, issues = discover_resources(cfg, folder_filter)

    n_invalid = sum(1 for r in issues if r.status == "invalid")
    n_dupes = sum(1 for r in issues if r.status == "skipped")
    click.echo(f"  Found {len(resources)} valid resource(s) to process.")
    if n_invalid:
        click.echo(f"  {n_invalid} invalid file(s) will be skipped.", err=True)
    if n_dupes:
        click.echo(f"  {n_dupes} duplicate(s) will be skipped.", err=True)

    if not resources:
        if issues:
            sys.exit(print_summary([], issues, verbose))
        click.echo("No resources found.")
        sys.exit(0)

    click.echo("Authenticating with Keycloak...")
    auth = AuthClient(cfg)
    auth.authenticate()
    click.echo("Authentication successful.")

    if dry_run:
        click.echo(f"\nDry-run mode -- no uploads will be performed.")
        dry_results = [
            UploadResult(folder=r.folder, resource=r.key, status="dry-run", file_path=r.rel_path)
            for r in resources
        ]
        sys.exit(print_summary(dry_results, issues, verbose))

    click.echo(f"\nUploading {len(resources)} resource(s)...")
    results: list[UploadResult] = []
    for i, resource in enumerate(resources, 1):
        result = upload_resource(resource, auth, cfg.fhir_base_url, verbose)
        results.append(result)
        icon = "[OK]" if result.status == "ok" else "[FAIL]"
        status_str = f"{result.code} {result.reason}" if result.code else result.reason
        click.echo(f"  [{i}/{len(resources)}] {icon} {resource.key} -> {status_str}")

    sys.exit(print_summary(results, issues, verbose))


if __name__ == "__main__":
    main()