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 
22
23
24
25
26
27
def __init__(
    self,
    store: Dataset,
):
    self.store: Dataset = store
    super().__init__(store)

store instance-attribute 

store: Dataset = store

serialize 

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

Serialize the store to the given stream.

Parameters:

  • stream

    (IO[bytes]) –

    The stream to serialize to.

  • base

    (str | None, default: None ) –

    The base URI to use for the serialization.

  • encoding

    (str | None, 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 
29
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
def serialize(
    self,
    stream: IO[bytes],
    base: str | None = None,
    encoding: str | None = 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")
    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():
        if header_id:
            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"))