onconova.interoperability.controllers

ConflictResolution

Bases: str, Enum

Enum representing strategies for resolving conflicts during interoperability operations.

Attributes:

Name	Type	Description
OVERWRITE		Overwrite existing data with new data.
REASSIGN		Reassign ownership or association to resolve the conflict.

OVERWRITE class-attribute instance-attribute

REASSIGN class-attribute instance-attribute

ConflictingCaseException

Bases: APIException

Exception raised when an unresolved import case conflict occurs.

default_detail class-attribute instance-attribute

status_code class-attribute instance-attribute

ConflictingClinicalIdentifierException

Bases: APIException

Exception raised when a clinical identifier conflicts with an existing case.

default_detail class-attribute instance-attribute

status_code class-attribute instance-attribute

InteroperabilityController

Bases: ControllerBase

API controller for interoperability operations related to resource and patient case management.

export_case_bundle(caseId)

Exports a patient case bundle by retrieving the PatientCase object with the given case ID, creates an export event for the case, and returns the exported case object.

Source code in onconova/interoperability/controllers.py

@route.get(
    path="/bundles/{caseId}",
    response={
        200: PatientCaseBundle,
        **COMMON_HTTP_ERRORS,
    },
    permissions=[perms.CanExportData],
    operation_id="exportPatientCaseBundle",
)
def export_case_bundle(self, caseId: str):
    """
    Exports a patient case bundle by retrieving the PatientCase object with the given case ID,
    creates an export event for the case, and returns the exported case object.
    """
    exported_case = get_object_or_404(PatientCase, id=caseId)
    pghistory.create_event(exported_case, label="export")
    return exported_case

export_resource(resourceId)

Exports a resource identified by its UUID, serializing its data and associated metadata.

Notes:

- Creates an export event for the resource using `pghistory.create_event`.
- Computes a checksum of the exported data for integrity verification.

Source code in onconova/interoperability/controllers.py

@route.get(
    path="resources/{resourceId}",
    response={
        200: Any,
        404: None,
        **COMMON_HTTP_ERRORS,
    },
    permissions=[perms.CanManageCases, perms.CanExportData],
    operation_id="exportResource",
)
def export_resource(self, resourceId: str):
    """
    Exports a resource identified by its UUID, serializing its data and associated metadata.

    Notes:

        - Creates an export event for the resource using `pghistory.create_event`.
        - Computes a checksum of the exported data for integrity verification.

    """
    instance = find_uuid_across_models(resourceId)
    if not instance:
        return 404, None
    schema = next(
        (
            schema
            for schema in oncology_schemas.ONCOLOGY_SCHEMAS
            if getattr(schema, "__orm_model__", None) == instance._meta.model
        )
    )
    export_data = schema.model_validate(instance).model_dump(mode="json")
    metadata = ExportMetadata(
        exportedAt=datetime.now(),
        exportedBy=self.context.request.user.username,
        exportVersion=settings.VERSION,
        checksum=hashlib.md5(
            json.dumps(export_data, sort_keys=True).encode("utf-8")
        ).hexdigest(),
    ).model_dump(mode="json")
    pghistory.create_event(instance, label="export")
    return {**metadata, **export_data}

import_case_bundle(bundle, conflict=None)

Imports a patient case bundle into the database, handling conflicts based on the specified resolution strategy.

Source code in onconova/interoperability/controllers.py

@route.post(
    path="/bundles",
    response={
        201: ModifiedResourceSchema,
        422: None,
        **COMMON_HTTP_ERRORS,
    },
    permissions=[perms.CanManageCases],
    operation_id="importPatientCaseBundle",
)
def import_case_bundle(
    self, bundle: PatientCaseBundle, conflict: ConflictResolution = None
):
    """
    Imports a patient case bundle into the database, handling conflicts based on the specified resolution strategy.
    """

    conflicting_case = PatientCase.objects.filter(
        pseudoidentifier=bundle.pseudoidentifier
    ).first()
    with transaction.atomic():
        if conflicting_case:
            if not conflict:
                raise ConflictingCaseException()
            if conflict == ConflictResolution.OVERWRITE:
                caseId = conflicting_case.id
                conflicting_case.delete()
                conflicting_case = PatientCase(pk=caseId)
            elif conflict == ConflictResolution.REASSIGN:
                bundle.pseudoidentifier = ""
                conflicting_case = None
        if not conflicting_case:
            if PatientCase.objects.filter(
                clinical_identifier=bundle.clinicalIdentifier,
                clinical_center=bundle.clinicalCenter,
            ).exists():
                raise ConflictingClinicalIdentifierException()
        imported_case = BundleParser(bundle).import_bundle(case=conflicting_case)
        pghistory.create_event(imported_case, label="import")
    return 201, imported_case

resolve_resource_id(resourceId)

Resolves a resource ID by searching across models for a matching UUID.

Source code in onconova/interoperability/controllers.py

@route.get(
    path="resources/{resourceId}/description",
    response={
        200: str,
        404: None,
        **COMMON_HTTP_ERRORS,
    },
    permissions=[perms.CanViewCases],
    operation_id="resolveResourceId",
)
def resolve_resource_id(self, resourceId: str):
    """
    Resolves a resource ID by searching across models for a matching UUID.
    """
    instance = find_uuid_across_models(resourceId)
    if not instance:
        return 404, None
    return instance.description

2025-10-172025-10-17runner