FHIR Client

fhir_client.py

FHIR client wrapper for connecting to HAPI FHIR server. Provides utility functions for creating and reading FHIR resources including Patient demographics and Communication (messaging).

FhirCommunicationError

Bases: Exception

Raised when a FHIR Communication operation fails.

Source code in app/fhir_client.py

class FhirCommunicationError(Exception):
    """Raised when a FHIR Communication operation fails."""

get_fhir_client

get_fhir_client()

Get configured FHIR client instance.

Returns:

Name	Type	Description
FHIRClient	FHIRClient	Configured client connected to HAPI FHIR server.

Raises:

Type	Description
FhirCommunicationError	If clinical services are disabled.

Source code in app/fhir_client.py

def get_fhir_client() -> client.FHIRClient:
    """Get configured FHIR client instance.

    Returns:
        FHIRClient: Configured client connected to HAPI FHIR server.

    Raises:
        FhirCommunicationError: If clinical services are disabled.
    """
    _require_clinical_services()
    fhir_settings = {
        "app_id": "quill_medical",
        "api_base": settings.FHIR_SERVER_URL,
    }
    return client.FHIRClient(settings=fhir_settings)

add_avatar_gradient_extension

add_avatar_gradient_extension(patient, gradient_index=None)

Add avatar gradient index extension to FHIR Patient.

Stores a single gradient index (0-29) which maps to predefined gradients in the frontend.

Parameters:

Name	Type	Description	Default
patient	Patient	FHIR Patient resource instance.	required
gradient_index	int | None	Index of gradient (0-29), or None to generate random.	None

Example

patient = Patient() add_avatar_gradient_extension(patient, gradient_index=5)

patient.extension now contains gradientIndex = 5

Source code in app/fhir_client.py

def add_avatar_gradient_extension(
    patient: Patient, gradient_index: int | None = None
) -> None:
    """Add avatar gradient index extension to FHIR Patient.

    Stores a single gradient index (0-29) which maps to predefined
    gradients in the frontend.

    Args:
        patient: FHIR Patient resource instance.
        gradient_index: Index of gradient (0-29), or None to generate random.

    Example:
        >>> patient = Patient()
        >>> add_avatar_gradient_extension(patient, gradient_index=5)
        >>> # patient.extension now contains gradientIndex = 5
    """
    if gradient_index is None:
        gradient_index = generate_avatar_gradient_index()

    # Create extension with gradient index
    gradient_ext = Extension()
    gradient_ext.url = AVATAR_GRADIENT_EXTENSION_URL
    gradient_ext.valueInteger = gradient_index

    # Add to patient extensions
    if patient.extension is None:
        patient.extension = []
    patient.extension.append(gradient_ext)

create_fhir_patient

create_fhir_patient(given_name, family_name, birth_date=None, gender=None, nhs_number=None, mrn=None, patient_id=None)

Create a new FHIR Patient resource.

Parameters:

Name	Type	Description	Default
given_name	str	Patient's first/given name.	required
family_name	str	Patient's family/last name.	required
birth_date	str | None	Date of birth in YYYY-MM-DD format.	None
gender	str | None	Gender (male, female, other, unknown).	None
nhs_number	str | None	NHS number (10-digit UK national identifier).	None
mrn	str | None	Medical Record Number (local hospital identifier).	None
patient_id	str | None	Optional ID for the patient.	None

Returns:

Name	Type	Description
dict	dict[str, Any]	Created patient resource as dictionary.

Raises:

Type	Description
ValueError	If required fields are empty or invalid.
Exception	If creation fails.

Source code in app/fhir_client.py

def create_fhir_patient(
    given_name: str,
    family_name: str,
    birth_date: str | None = None,
    gender: str | None = None,
    nhs_number: str | None = None,
    mrn: str | None = None,
    patient_id: str | None = None,
) -> dict[str, Any]:
    """Create a new FHIR Patient resource.

    Args:
        given_name (str): Patient's first/given name.
        family_name (str): Patient's family/last name.
        birth_date (str | None): Date of birth in YYYY-MM-DD format.
        gender (str | None): Gender (male, female, other, unknown).
        nhs_number (str | None): NHS number (10-digit UK national identifier).
        mrn (str | None): Medical Record Number (local hospital identifier).
        patient_id (str | None): Optional ID for the patient.

    Returns:
        dict: Created patient resource as dictionary.

    Raises:
        ValueError: If required fields are empty or invalid.
        Exception: If creation fails.
    """
    # Defensive programming: validate required inputs
    if not given_name or not given_name.strip():
        raise ValueError("Given name cannot be empty")
    if not family_name or not family_name.strip():
        raise ValueError("Family name cannot be empty")

    # Validate gender if provided
    if gender is not None:
        valid_genders = {"male", "female", "other", "unknown"}
        if gender.lower() not in valid_genders:
            raise ValueError(
                f"Gender must be one of {valid_genders}, got: {gender}"
            )

    # Validate NHS number format if provided
    if nhs_number is not None and nhs_number.strip():
        cleaned_nhs = nhs_number.strip().replace(" ", "")
        if not cleaned_nhs.isdigit() or len(cleaned_nhs) != 10:
            raise ValueError(
                f"NHS number must be 10 digits, got: {nhs_number}"
            )
    fhir = get_fhir_client()

    # Create Patient resource
    patient = Patient()

    # Generate UUID for client-assigned ID (standard FHIR pattern)
    # FHIR servers must support client-assigned IDs via PUT requests
    patient_uuid = patient_id if patient_id else str(uuid.uuid4())
    patient.id = patient_uuid

    # Create name
    name = HumanName()
    name.given = [given_name]
    name.family = family_name
    name.use = "official"

    patient.name = [name]

    # Add birth date if provided
    if birth_date:
        patient.birthDate = FHIRDate(birth_date)

    # Add gender if provided (must be one of: male, female, other, unknown)
    if gender and gender.lower() in ["male", "female", "other", "unknown"]:
        patient.gender = gender.lower()

    # Add identifiers
    identifiers = []

    # NHS Number (UK national identifier)
    if nhs_number:
        from fhirclient.models.identifier import (  # type: ignore[import-untyped]
            Identifier,
        )

        nhs_id = Identifier()
        nhs_id.system = "https://fhir.nhs.uk/Id/nhs-number"
        nhs_id.value = nhs_number
        identifiers.append(nhs_id)

    # Medical Record Number (local hospital identifier)
    if mrn:
        from fhirclient.models.identifier import Identifier

        mrn_id = Identifier()
        mrn_id.system = "http://hospital.example.org/identifiers/mrn"
        mrn_id.value = mrn
        identifiers.append(mrn_id)

    if identifiers:
        patient.identifier = identifiers

    # Add avatar gradient extension (generate colors automatically)
    add_avatar_gradient_extension(patient)

    # Use PUT to create with client-assigned UUID (standard FHIR pattern)
    # PUT /Patient/{uuid} creates the resource with our specified ID
    # This is standard FHIR behavior - no server configuration needed
    fhir.server.put_json(f"Patient/{patient_uuid}", patient.as_json())

    # Return the patient data with the UUID
    return patient.as_json()  # type: ignore[no-any-return]

read_fhir_patient

read_fhir_patient(patient_id)

Read a FHIR Patient resource by ID.

Parameters:

Name	Type	Description	Default
patient_id	str	FHIR Patient resource ID.	required

Returns:

Type	Description
dict[str, Any] | None	dict | None: Patient resource as dictionary, or None if not found.

Source code in app/fhir_client.py

def read_fhir_patient(patient_id: str) -> dict[str, Any] | None:
    """Read a FHIR Patient resource by ID.

    Args:
        patient_id (str): FHIR Patient resource ID.

    Returns:
        dict | None: Patient resource as dictionary, or None if not found.
    """
    fhir = get_fhir_client()

    try:
        patient = Patient.read(patient_id, fhir.server)
        return patient.as_json()  # type: ignore[no-any-return]
    except Exception:
        return None

delete_fhir_patient

delete_fhir_patient(patient_id)

Delete a FHIR Patient resource by ID.

WARNING: This function is ONLY for development/testing purposes! It should NOT be exposed via API endpoints in production. Patient data should be soft-deleted (Patient.active = false) instead.

Parameters:

Name	Type	Description	Default
patient_id	str	FHIR Patient resource ID to delete.	required

Returns:

Name	Type	Description
bool	bool	True if deletion successful, False otherwise.

Source code in app/fhir_client.py

def delete_fhir_patient(patient_id: str) -> bool:
    """Delete a FHIR Patient resource by ID.

    WARNING: This function is ONLY for development/testing purposes!
    It should NOT be exposed via API endpoints in production.
    Patient data should be soft-deleted (Patient.active = false) instead.

    Args:
        patient_id (str): FHIR Patient resource ID to delete.

    Returns:
        bool: True if deletion successful, False otherwise.
    """
    fhir = get_fhir_client()

    try:
        patient = Patient.read(patient_id, fhir.server)
        patient.delete()
        return True
    except Exception:
        return False

list_fhir_patients

list_fhir_patients()

List all FHIR Patient resources.

Returns:

Type	Description
list[dict[str, Any]]	list[dict]: List of patient resources as dictionaries.

Source code in app/fhir_client.py

def list_fhir_patients() -> list[dict[str, Any]]:
    """List all FHIR Patient resources.

    Returns:
        list[dict]: List of patient resources as dictionaries.
    """
    fhir = get_fhir_client()

    try:
        # Search for all patients
        search = Patient.where(struct={})
        patients = search.perform_resources(fhir.server)
        return [p.as_json() for p in patients]
    except Exception:
        return []

update_fhir_patient

update_fhir_patient(patient_id, demographics)

Update a FHIR Patient resource with demographics data.

Parameters:

Name	Type	Description	Default
patient_id	str	FHIR Patient resource ID.	required
demographics	dict	Dictionary containing demographics fields: - given_name (str): First/given name - family_name (str): Family/last name - date_of_birth (str): ISO date string - sex (str): Gender (male|female|other|unknown) - address (dict): Address information - contact (dict): Contact information	required

Returns:

Type	Description
dict[str, Any] | None	dict | None: Updated patient resource as dictionary, or None if not found.

Raises:

Type	Description
Exception	If update fails.

Source code in app/fhir_client.py

def update_fhir_patient(
    patient_id: str, demographics: dict[str, Any]
) -> dict[str, Any] | None:
    """Update a FHIR Patient resource with demographics data.

    Args:
        patient_id (str): FHIR Patient resource ID.
        demographics (dict): Dictionary containing demographics fields:
            - given_name (str): First/given name
            - family_name (str): Family/last name
            - date_of_birth (str): ISO date string
            - sex (str): Gender (male|female|other|unknown)
            - address (dict): Address information
            - contact (dict): Contact information

    Returns:
        dict | None: Updated patient resource as dictionary, or None if not found.

    Raises:
        Exception: If update fails.
    """
    fhir = get_fhir_client()

    try:
        # Read existing patient
        patient = Patient.read(patient_id, fhir.server)

        # Update name if provided
        if demographics.get("given_name") or demographics.get("family_name"):
            name = HumanName()
            name.given = [demographics.get("given_name", "")]
            name.family = demographics.get("family_name", "")
            name.use = "official"
            patient.name = [name]

        # Update birth date if provided
        if demographics.get("date_of_birth"):
            patient.birthDate = FHIRDate(demographics["date_of_birth"])

        # Update gender if provided
        if demographics.get("sex"):
            # Map to FHIR gender values
            gender_map = {
                "male": "male",
                "female": "female",
                "other": "other",
                "unknown": "unknown",
            }
            patient.gender = gender_map.get(
                demographics["sex"].lower(), "unknown"
            )

        # Update address if provided
        if demographics.get("address"):
            addr = Address()
            addr_data = demographics["address"]
            if addr_data.get("line"):
                addr.line = addr_data["line"]
            if addr_data.get("city"):
                addr.city = addr_data["city"]
            if addr_data.get("state"):
                addr.state = addr_data["state"]
            if addr_data.get("postalCode"):
                addr.postalCode = addr_data["postalCode"]
            if addr_data.get("country"):
                addr.country = addr_data["country"]
            patient.address = [addr]

        # Update contact/telecom if provided
        if demographics.get("contact"):
            contact_data = demographics["contact"]
            telecoms = []

            if contact_data.get("phone"):
                phone = ContactPoint()
                phone.system = "phone"
                phone.value = contact_data["phone"]
                telecoms.append(phone)

            if contact_data.get("email"):
                email = ContactPoint()
                email.system = "email"
                email.value = contact_data["email"]
                telecoms.append(email)

            if telecoms:
                patient.telecom = telecoms

        # Save updates
        patient.update(fhir.server)

        return patient.as_json()  # type: ignore[no-any-return]
    except Exception:
        return None

create_fhir_communication

create_fhir_communication(*, conversation_id, patient_id, sender_display, sender_user_id, body, first_message_fhir_id=None, amends_fhir_id=None)

Create a FHIR Communication resource for a message.

Parameters:

Name	Type	Description	Default
conversation_id	str	UUID grouping messages into a thread.	required
patient_id	str	FHIR Patient resource ID the conversation is about.	required
sender_display	str	Human-readable sender name.	required
sender_user_id	int	Auth DB user ID of the sender.	required
body	str	Message text content.	required
first_message_fhir_id	str | None	FHIR ID of the first message in the thread (for partOf linking). None for the first message.	None
amends_fhir_id	str | None	FHIR ID of a message this one amends.	None

Returns:

Name	Type	Description
dict	dict[str, Any]	Created Communication resource as JSON dict.

Raises:

Type	Description
FhirCommunicationError	If creation fails.

Source code in app/fhir_client.py

def create_fhir_communication(
    *,
    conversation_id: str,
    patient_id: str,
    sender_display: str,
    sender_user_id: int,
    body: str,
    first_message_fhir_id: str | None = None,
    amends_fhir_id: str | None = None,
) -> dict[str, Any]:
    """Create a FHIR Communication resource for a message.

    Args:
        conversation_id: UUID grouping messages into a thread.
        patient_id: FHIR Patient resource ID the conversation is about.
        sender_display: Human-readable sender name.
        sender_user_id: Auth DB user ID of the sender.
        body: Message text content.
        first_message_fhir_id: FHIR ID of the first message in the
            thread (for partOf linking). None for the first message.
        amends_fhir_id: FHIR ID of a message this one amends.

    Returns:
        dict: Created Communication resource as JSON dict.

    Raises:
        FhirCommunicationError: If creation fails.
    """
    fhir = get_fhir_client()
    comm_uuid = str(uuid.uuid4())

    resource: dict[str, Any] = {
        "resourceType": "Communication",
        "id": comm_uuid,
        "status": "completed",
        "subject": {"reference": f"Patient/{patient_id}"},
        "sender": {
            "display": sender_display,
            "extension": [
                {
                    "url": SENDER_USER_ID_EXTENSION_URL,
                    "valueInteger": sender_user_id,
                }
            ],
        },
        "sent": FHIRDateTime(datetime_module_now_utc_iso()).as_json(),
        "payload": [{"contentString": body}],
        "extension": [
            {
                "url": CONVERSATION_ID_EXTENSION_URL,
                "valueString": conversation_id,
            }
        ],
    }

    if first_message_fhir_id:
        resource["partOf"] = [
            {"reference": f"Communication/{first_message_fhir_id}"}
        ]

    if amends_fhir_id:
        resource["extension"].append(
            {
                "url": AMENDS_EXTENSION_URL,
                "valueReference": {
                    "reference": f"Communication/{amends_fhir_id}",
                },
            }
        )

    try:
        fhir.server.put_json(f"Communication/{comm_uuid}", resource)
    except Exception as exc:
        logger.error(
            "Failed to create FHIR Communication %s: %s",
            comm_uuid,
            exc,
        )
        raise FhirCommunicationError(
            f"Failed to store message in FHIR: {exc}"
        ) from exc

    resource["id"] = comm_uuid
    return resource

datetime_module_now_utc_iso

datetime_module_now_utc_iso()

Return current UTC time as ISO-8601 string.

Source code in app/fhir_client.py

def datetime_module_now_utc_iso() -> str:
    """Return current UTC time as ISO-8601 string."""
    from datetime import UTC, datetime

    return datetime.now(UTC).isoformat()