Skip to content

# `` patch

Classes:

- `Operation` –

  Enum of RDF Patch operations.

- `RDFPatchParser` –

## `` __all__ `module-attribute`

    __all__ = ['RDFPatchParser', 'Operation']

## `` Operation

Bases: `Enum`

Enum of RDF Patch operations.

Operations: - `AddTripleOrQuad` (A): Adds a triple or quad. - `DeleteTripleOrQuad` (D): Deletes a triple or quad. - `AddPrefix` (PA): Adds a prefix. - `DeletePrefix` (PD): Deletes a prefix. - `TransactionStart` (TX): Starts a transaction. - `TransactionCommit` (TC): Commits a transaction. - `TransactionAbort` (TA): Aborts a transaction. - `Header` (H): Specifies a header.

Attributes:

- `AddPrefix` –
- `AddTripleOrQuad` –
- `DeletePrefix` –
- `DeleteTripleOrQuad` –
- `Header` –
- `TransactionAbort` –
- `TransactionCommit` –
- `TransactionStart` –

### `` AddPrefix `class-attribute` `instance-attribute`

    AddPrefix = 'PA'

### `` AddTripleOrQuad `class-attribute` `instance-attribute`

    AddTripleOrQuad = 'A'

### `` DeletePrefix `class-attribute` `instance-attribute`

    DeletePrefix = 'PD'

### `` DeleteTripleOrQuad `class-attribute` `instance-attribute`

    DeleteTripleOrQuad = 'D'

### `` Header `class-attribute` `instance-attribute`

    Header = 'H'

### `` TransactionAbort `class-attribute` `instance-attribute`

    TransactionAbort = 'TA'

### `` TransactionCommit `class-attribute` `instance-attribute`

    TransactionCommit = 'TC'

### `` TransactionStart `class-attribute` `instance-attribute`

    TransactionStart = 'TX'

## `` RDFPatchParser

    RDFPatchParser(sink: Optional[Union[DummySink, NTGraphSink]] = None, bnode_context: Optional[_BNodeContextType] = None)

Bases: `NQuadsParser`

Methods:

- `add_or_remove_triple_or_quad` –
- `add_prefix` –
- `delete_prefix` –
- `eat_op` –
- `labeled_bnode` –
- `nodeid` –
- `operation` –
- `parse` –

  Parse inputsource as an RDF Patch file.

- `parsepatch` –

Source code in `rdflib/plugins/parsers/ntriples.py`

    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166

    def __init__(
        self,
        sink: Optional[Union[DummySink, NTGraphSink]] = None,
        bnode_context: Optional[_BNodeContextType] = None,
    ):
        self.skolemize = False

        if bnode_context is not None:
            self._bnode_ids = bnode_context
        else:
            self._bnode_ids = {}

        self.sink: Union[DummySink, NTGraphSink]
        if sink is not None:
            self.sink = sink
        else:
            self.sink = DummySink()

        self.buffer: Optional[str] = None
        self.file: Optional[Union[TextIO, codecs.StreamReader]] = None
        self.line: Optional[str] = ""

### `` add_or_remove_triple_or_quad

    add_or_remove_triple_or_quad(operation, bnode_context: Optional[_BNodeContextType] = None) -> None

Source code in `rdflib/plugins/parsers/patch.py`

    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142

    def add_or_remove_triple_or_quad(
        self, operation, bnode_context: Optional[_BNodeContextType] = None
    ) -> None:
        self.eat(r_wspace)
        if (not self.line) or self.line.startswith("#"):
            return  # The line is empty or a comment

        subject = self.labeled_bnode() or self.subject(bnode_context)
        self.eat(r_wspace)

        predicate = self.predicate()
        self.eat(r_wspace)

        obj = self.labeled_bnode() or self.object(bnode_context)
        self.eat(r_wspace)

        context = self.labeled_bnode() or self.uriref() or self.nodeid(bnode_context)
        self.eat(r_tail)

        if self.line:
            raise ParseError("Trailing garbage")
        # Must have a context aware store - add on a normal Graph
        # discards anything where the ctx != graph.identifier
        if operation == Operation.AddTripleOrQuad:
            if context:
                self.sink.get_context(context).add((subject, predicate, obj))
            else:
                self.sink.default_context.add((subject, predicate, obj))
        elif operation == Operation.DeleteTripleOrQuad:
            if context:
                self.sink.get_context(context).remove((subject, predicate, obj))
            else:
                self.sink.default_context.remove((subject, predicate, obj))

### `` add_prefix

    add_prefix()

Source code in `rdflib/plugins/parsers/patch.py`

    144
    145
    146
    147
    148

    def add_prefix(self):
        # Extract prefix and URI from the line
        prefix, ns, _ = self.line.replace('"', "").replace("'", "").split(" ")  # type: ignore[union-attr]
        ns_stripped = ns.strip("<>")
        self.sink.bind(prefix, ns_stripped)

### `` delete_prefix

    delete_prefix()

Source code in `rdflib/plugins/parsers/patch.py`

    150
    151
    152

    def delete_prefix(self):
        prefix, _, _ = self.line.replace('"', "").replace("'", "").split(" ")  # type: ignore[union-attr]
        self.sink.namespace_manager.bind(prefix, None, replace=True)

### `` eat_op

    eat_op(op: str) -> None

Source code in `rdflib/plugins/parsers/patch.py`

    164
    165

    def eat_op(self, op: str) -> None:
        self.line = self.line.lstrip(op)  # type: ignore[union-attr]

### `` labeled_bnode

    labeled_bnode()

Source code in `rdflib/plugins/parsers/patch.py`

    174
    175
    176
    177
    178
    179

    def labeled_bnode(self):
        if self.peek("<_"):
            plain_uri = self.eat(r_uriref).group(1)
            bnode_id = r_nodeid.match(plain_uri).group(1)  # type: ignore[union-attr]
            return BNode(bnode_id)
        return False

### `` nodeid

    nodeid(bnode_context: Optional[_BNodeContextType] = None) -> Union[Literal[False], BNode, URIRef]

Source code in `rdflib/plugins/parsers/patch.py`

    167
    168
    169
    170
    171
    172

    def nodeid(
        self, bnode_context: Optional[_BNodeContextType] = None
    ) -> Union[te.Literal[False], BNode, URIRef]:
        if self.peek("_"):
            return BNode(self.eat(r_nodeid).group(1))
        return False

### `` operation

    operation() -> Operation

Source code in `rdflib/plugins/parsers/patch.py`

    154
    155
    156
    157
    158
    159
    160
    161
    162

    def operation(self) -> Operation:
        for op in Operation:
            if self.line.startswith(op.value):  # type: ignore[union-attr]
                self.eat_op(op.value)
                return op
        raise ValueError(
            f'Invalid or no Operation found in line: "{self.line}". Valid Operations '
            f"codes are {', '.join([op.value for op in Operation])}"
        )

### `` parse

    parse(inputsource: InputSource, sink: Dataset, bnode_context: Optional[_BNodeContextType] = None, skolemize: bool = False, **kwargs: Any) -> Dataset

Parse inputsource as an RDF Patch file.

Parameters:

- #### `inputsource`

  (`InputSource`) –

  the source of RDF Patch formatted data

- #### `sink`

  (`Dataset`) –

  where to send parsed data

- #### `bnode_context`

  (`Optional[_BNodeContextType]`, default: `None` ) –

  a dict mapping blank node identifiers to `BNode` instances. See `.W3CNTriplesParser.parse`

Source code in `rdflib/plugins/parsers/patch.py`

    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

    def parse(  # type: ignore[override]
        self,
        inputsource: InputSource,
        sink: Dataset,
        bnode_context: Optional[_BNodeContextType] = None,
        skolemize: bool = False,
        **kwargs: Any,
    ) -> Dataset:
        """Parse inputsource as an RDF Patch file.

        Args:
            inputsource: the source of RDF Patch formatted data
            sink: where to send parsed data
            bnode_context: a dict mapping blank node identifiers to [`BNode`][rdflib.term.BNode]
                instances. See `.W3CNTriplesParser.parse`
        """
        assert sink.store.context_aware, (
            "RDFPatchParser must be given" " a context aware store."
        )
        # type error: Incompatible types in assignment (expression has type "ConjunctiveGraph", base class "W3CNTriplesParser" defined the type as "Union[DummySink, NTGraphSink]")
        self.sink: Dataset = Dataset(store=sink.store)
        self.skolemize = skolemize

        source = inputsource.getCharacterStream()
        if not source:
            source = inputsource.getByteStream()
            source = getreader("utf-8")(source)

        if not hasattr(source, "read"):
            raise ParseError("Item to parse must be a file-like object.")

        self.file = source
        self.buffer = ""
        while True:
            self.line = __line = self.readline()
            if self.line is None:
                break
            try:
                self.parsepatch(bnode_context)
            except ParseError as msg:
                raise ParseError("Invalid line (%s):\n%r" % (msg, __line))
        return self.sink

### `` parsepatch

    parsepatch(bnode_context: Optional[_BNodeContextType] = None) -> None

Source code in `rdflib/plugins/parsers/patch.py`

     92
     93
     94
     95
     96
     97
     98
     99
    100
    101
    102
    103
    104
    105
    106
    107
    108

    def parsepatch(self, bnode_context: Optional[_BNodeContextType] = None) -> None:
        self.eat(r_wspace)
        #  From spec: "No comments should be included (comments start # and run to end
        #  of line)."
        if (not self.line) or self.line.startswith("#"):
            return  # The line is empty or a comment

        # if header, transaction, skip
        operation = self.operation()
        self.eat(r_wspace)

        if operation in [Operation.AddTripleOrQuad, Operation.DeleteTripleOrQuad]:
            self.add_or_remove_triple_or_quad(operation, bnode_context)
        elif operation == Operation.AddPrefix:
            self.add_prefix()
        elif operation == Operation.DeletePrefix:
            self.delete_prefix()