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

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

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

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

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

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

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

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

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

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

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