Skip to content

# `` patch

Classes:

- `PatchSerializer` –

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

Attributes:

- `add_remove_methods` –

## `` 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:

- `store` (`Dataset`) –

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"))