Skip to content

Event synchronisation

kgm.event.syncer

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:

Name Type Description Default
current_working_directory Path

The current working directory path.

required
manifest Path | tuple[Path, Path, Graph]

The path of the KGM file to be loaded.

required
sparql_endpoint str

The URL of the SPARQL Endpoint.

required
http_client Client

The HTTP client to use for making requests.

required
event_client EventClient

The event client to use for sending events.

required
Source code in kgm/event/syncer.py
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
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)