Skip to content

patch

Classes:

  • PatchSerializer –

    Creates an RDF patch file to add and remove triples/quads.

Attributes:

add_remove_methods module-attribute 

add_remove_methods = {'add': 'A', 'remove': 'D'}

PatchSerializer 

PatchSerializer(store: Dataset)

Bases: Serializer

Creates an RDF patch file to add and remove triples/quads. Can either: - Create an add or delete patch for a single Dataset. - Create a patch to represent the difference between two Datasets.

Methods:

  • serialize –

    Serialize the store to the given stream.

Attributes:

Source code in rdflib/plugins/serializers/patch.py 
23
24
25
26
27
28
def __init__(
    self,
    store: Dataset,
):
    self.store: Dataset = store
    super().__init__(store)

store instance-attribute 

store: Dataset = store

serialize 

serialize(stream: IO[bytes], base: Optional[str] = None, encoding: Optional[str] = None, **kwargs: Any) -> None

Serialize the store to the given stream.

Parameters:

  • stream

    (IO[bytes]) –

    The stream to serialize to.

  • base

    (Optional[str], default: None ) –

    The base URI to use for the serialization.

  • encoding

    (Optional[str], default: None ) –

    The encoding to use for the serialization.

  • kwargs

    (Any, default: {} ) –

    Additional keyword arguments.

Supported keyword arguments:

  • operation: The operation to perform. Either ‘add’ or ‘remove’.
  • target: The target Dataset to compare against. NB: Only one of ‘operation’ or ‘target’ should be provided.
  • header_id: The header ID to use.
  • header_prev: The previous header ID to use.
Source code in rdflib/plugins/serializers/patch.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def serialize(
    self,
    stream: IO[bytes],
    base: Optional[str] = None,
    encoding: Optional[str] = None,
    **kwargs: Any,
) -> None:
    """Serialize the store to the given stream.

    Args:
        stream: The stream to serialize to.
        base: The base URI to use for the serialization.
        encoding: The encoding to use for the serialization.
        kwargs: Additional keyword arguments.

    Supported keyword arguments:

    - operation: The operation to perform. Either 'add' or 'remove'.
    - target: The target Dataset to compare against.
    NB: Only one of 'operation' or 'target' should be provided.
    - header_id: The header ID to use.
    - header_prev: The previous header ID to use.
    """
    operation = kwargs.get("operation")
    target = kwargs.get("target")
    header_id = kwargs.get("header_id")
    header_prev = kwargs.get("header_prev")
    if not header_id:
        header_id = f"uuid:{uuid4()}"
    encoding = self.encoding
    if base is not None:
        warnings.warn("PatchSerializer does not support base.")
    if encoding is not None and encoding.lower() != self.encoding.lower():
        warnings.warn(
            "PatchSerializer does not use custom encoding. "
            f"Given encoding was: {encoding}"
        )

    def write_header():
        stream.write(f"H id <{header_id}> .\n".encode(encoding, "replace"))
        if header_prev:
            stream.write(f"H prev <{header_prev}>\n".encode(encoding, "replace"))
        stream.write("TX .\n".encode(encoding, "replace"))

    def write_triples(contexts, op_code, use_passed_contexts=False):
        for context in contexts:
            if not use_passed_contexts:
                context = self.store.get_context(context.identifier)
            for triple in context:
                stream.write(
                    self._patch_row(triple, context.identifier, op_code).encode(
                        encoding, "replace"
                    )
                )

    if operation:
        assert operation in add_remove_methods, f"Invalid operation: {operation}"
    elif not target:
        # No operation specified and no target specified
        # Fall back to default operation of "add" to prevent a no-op
        operation = "add"
    write_header()
    if operation:
        operation_code = add_remove_methods.get(operation)
        write_triples(self.store.contexts(), operation_code)
    elif target:
        to_add, to_remove = self._diff(target)
        write_triples(to_add.contexts(), "A", use_passed_contexts=True)
        write_triples(to_remove.contexts(), "D", use_passed_contexts=True)

    stream.write("TC .\n".encode(encoding, "replace"))