Event synchronisation¶

def sync_rdf_delta(
    current_working_directory: Path,
    manifest: Path | tuple[Path, Path, Graph],
    sparql_endpoint: str,
    http_client: httpx.Client,
    event_client: EventClient,
):
    """Synchronize a KGM's resources with an event-based system that takes RDF patches.

    Parameters:
        current_working_directory: The current working directory path.
        manifest: The path of the KGM file to be loaded.
        sparql_endpoint: The URL of the SPARQL Endpoint.
        http_client: The HTTP client to use for making requests.
        event_client: The event client to use for sending events.
    """

    # Load the manifest on the latest commit.
    ds = load(manifest, return_data_type=ReturnDatatype.dataset)
    system_graph = ds.graph(OLIS.SystemGraph)
    vg_iri = system_graph.value(predicate=RDF.type, object=OLIS.VirtualGraph)
    if vg_iri is None:
        raise ValueError(
            "Could not find the Virtual Graph instance in the Olis system graph"
        )
    logger.info(f"Virtual Graph IRI: {vg_iri}")

    # Query the SPARQL endpoint and retrieve the git commit hash version from the system graph.
    previous_commit_hash = _retrieve_commit_hash(vg_iri, sparql_endpoint, http_client)
    logger.info(f"Previous commit hash: {previous_commit_hash}")

    # The current commit hash. Assume this is the latest.
    repo = Repo(current_working_directory)
    current_commit_hash = repo.head.commit.hexsha
    logger.info(f"Current commit hash: {current_commit_hash}")

    if previous_commit_hash is None:
        logger.info(
            "Previous commit hash is None. Adding current commit hash to dataset."
        )
        logger.info("Adding commit hash to current manifest dataset")
        _add_commit_hash_to_dataset(current_commit_hash, ds)
        logger.info("Generating RDF patch body chunks for add operation")
        rdf_patch_body_chunks = _generate_rdf_patch_body_add(ds)
    else:
        # Check out the previous commit.
        # Generate the previous manifest dataset.
        logger.info(f"Checking out previous commit: {previous_commit_hash}")
        repo.git.checkout(previous_commit_hash)
        logger.info("Loading previous manifest dataset")
        previous_ds = load(manifest, return_data_type=ReturnDatatype.dataset)
        logger.info("Adding commit hash to previous manifest dataset")
        _add_commit_hash_to_dataset(previous_commit_hash, previous_ds)
        logger.info("Adding commit hash to current manifest dataset")
        _add_commit_hash_to_dataset(current_commit_hash, ds)

        # Generate an RDF patch between the previous commit dataset and the current commit dataset.
        logger.info("Generating RDF patch body chunks for diff operation")
        rdf_patch_body_chunks = _generate_rdf_patch_body_diff(ds, previous_ds)

    # Create events for each chunk.
    for i, chunk in enumerate(rdf_patch_body_chunks):
        logger.info(f"Creating event for chunk {i + 1}")
        event_client.create_event(chunk)