Skip to content

# `` trig

Classes:

- `TrigParser` –

  An RDFLib parser for TriG

- `TrigSinkParser` –

Functions:

- `becauseSubGraph` –

## `` TrigParser

    TrigParser()

Bases: `Parser`

An RDFLib parser for TriG

Methods:

- `parse` –

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

    141
    142

    def __init__(self):
        pass

### `` parse

    parse(source: InputSource, graph: Graph, encoding: str = 'utf-8') -> None

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

    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175

    def parse(self, source: InputSource, graph: Graph, encoding: str = "utf-8") -> None:
        if encoding not in [None, "utf-8"]:
            raise Exception(
                # type error: Unsupported left operand type for % ("Tuple[str, str]")
                ("TriG files are always utf-8 encoded, ", "I was passed: %s")  # type: ignore[operator]
                % encoding
            )

        # we're currently being handed a Graph, not a ConjunctiveGraph
        assert graph.store.context_aware, "TriG Parser needs a context-aware store!"

        conj_graph = ConjunctiveGraph(store=graph.store, identifier=graph.identifier)
        conj_graph.default_context = graph  # TODO: CG __init__ should have a
        # default_context arg
        # TODO: update N3Processor so that it can use conj_graph as the sink
        conj_graph.namespace_manager = graph.namespace_manager

        sink = RDFSink(conj_graph)

        baseURI = conj_graph.absolutize(  # noqa: N806
            source.getPublicId() or source.getSystemId() or ""
        )
        p = TrigSinkParser(sink, baseURI=baseURI, turtle=True)

        stream = source.getCharacterStream()  # try to get str stream first
        if not stream:
            # fallback to get the bytes stream
            stream = source.getByteStream()
        p.loadStream(stream)

        for prefix, namespace in p._bindings.items():
            conj_graph.bind(prefix, namespace)

## `` TrigSinkParser

    TrigSinkParser(store: RDFSink, openFormula: Optional[Formula] = None, thisDoc: str = '', baseURI: Optional[str] = None, genPrefix: str = '', why: Optional[Callable[[], None]] = None, turtle: bool = False)

Bases: `SinkParser`

Methods:

- `directiveOrStatement` –
- `graph` –

  Parse trig graph, i.e.

- `labelOrSubject` –

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

    397
    398
    399
    400
    401
    402
    403
    404
    405
    406
    407
    408
    409
    410
    411
    412
    413
    414
    415
    416
    417
    418
    419
    420
    421
    422
    423
    424
    425
    426
    427
    428
    429
    430
    431
    432
    433
    434
    435
    436
    437
    438
    439
    440
    441
    442
    443
    444
    445
    446
    447
    448
    449
    450
    451
    452
    453
    454
    455
    456
    457
    458
    459
    460
    461
    462
    463
    464
    465
    466
    467
    468
    469
    470
    471
    472
    473

    def __init__(
        self,
        store: RDFSink,
        openFormula: Optional[Formula] = None,
        thisDoc: str = "",
        baseURI: Optional[str] = None,
        genPrefix: str = "",
        why: Optional[Callable[[], None]] = None,
        turtle: bool = False,
    ):
        """note: namespace names should *not* end in  # ;
        the  # will get added during qname processing"""

        self._bindings = {}
        if thisDoc != "":
            assert ":" in thisDoc, "Document URI not absolute: <%s>" % thisDoc
            self._bindings[""] = thisDoc + "#"  # default

        self._store = store
        if genPrefix:
            # TODO FIXME: there is no function named setGenPrefix
            store.setGenPrefix(genPrefix)  # type: ignore[attr-defined] # pass it on

        self._thisDoc = thisDoc
        self.lines = 0  # for error handling
        self.startOfLine = 0  # For calculating character number
        self._genPrefix = genPrefix
        self.keywords = ["a", "this", "bind", "has", "is", "of", "true", "false"]
        self.keywordsSet = 0  # Then only can others be considered qnames
        self._anonymousNodes: Dict[str, BNode] = {}
        # Dict of anon nodes already declared ln: Term
        self._variables: Dict[str, Variable] = {}
        self._parentVariables: Dict[str, Variable] = {}
        self._reason = why  # Why the parser was asked to parse this

        self.turtle = turtle  # raise exception when encountering N3 extensions
        # Turtle allows single or double quotes around strings, whereas N3
        # only allows double quotes.
        self.string_delimiters = ('"', "'") if turtle else ('"',)

        self._reason2: Optional[Callable[..., None]] = None  # Why these triples
        # was: diag.tracking
        if tracking:
            # type error: "BecauseOfData" does not return a value
            self._reason2 = BecauseOfData(  # type: ignore[func-returns-value]
                store.newSymbol(thisDoc), because=self._reason
            )

        self._baseURI: Optional[str]
        if baseURI:
            self._baseURI = baseURI
        else:
            if thisDoc:
                self._baseURI = thisDoc
            else:
                self._baseURI = None

        assert not self._baseURI or ":" in self._baseURI

        if not self._genPrefix:
            if self._thisDoc:
                self._genPrefix = self._thisDoc + "#_g"
            else:
                self._genPrefix = uniqueURI()

        self._formula: Optional[Formula]
        if openFormula is None and not turtle:
            if self._thisDoc:
                # TODO FIXME: store.newFormula does not take any arguments
                self._formula = store.newFormula(thisDoc + "#_formula")  # type: ignore[call-arg]
            else:
                self._formula = store.newFormula()
        else:
            self._formula = openFormula

        self._context: Optional[Formula] = self._formula
        self._parentContext: Optional[Formula] = None

### `` directiveOrStatement

    directiveOrStatement(argstr: str, h: int) -> int

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

    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39

    def directiveOrStatement(self, argstr: str, h: int) -> int:  # noqa: N802
        # import pdb; pdb.set_trace()

        i = self.skipSpace(argstr, h)
        if i < 0:
            return i  # EOF

        j = self.graph(argstr, i)
        if j >= 0:
            return j

        j = self.sparqlDirective(argstr, i)
        if j >= 0:
            return j

        j = self.directive(argstr, i)
        if j >= 0:
            return self.checkDot(argstr, j)

        j = self.statement(argstr, i)
        if j >= 0:
            return self.checkDot(argstr, j)

        return j

### `` graph

    graph(argstr: str, i: int) -> int

Parse trig graph, i.e.

= { .. triples .. }

return -1 if it doesn’t look like a graph-decl raise Exception if it looks like a graph, but isn’t.

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

     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
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132

    def graph(self, argstr: str, i: int) -> int:
        """
        Parse trig graph, i.e.

           <urn:graphname> = { .. triples .. }

        return -1 if it doesn't look like a graph-decl
        raise Exception if it looks like a graph, but isn't.
        """

        need_graphid = False
        # import pdb; pdb.set_trace()
        j = self.sparqlTok("GRAPH", argstr, i)  # optional GRAPH keyword
        if j >= 0:
            i = j
            need_graphid = True

        r: MutableSequence[Any] = []
        j = self.labelOrSubject(argstr, i, r)
        if j >= 0:
            graph = r[0]
            i = j
        elif need_graphid:
            self.BadSyntax(argstr, i, "GRAPH keyword must be followed by graph name")
        else:
            graph = self._store.graph.identifier  # hack

        j = self.skipSpace(argstr, i)
        if j < 0:
            self.BadSyntax(argstr, i, "EOF found when expected graph")

        if argstr[j : j + 1] == "=":  # optional = for legacy support
            i = self.skipSpace(argstr, j + 1)
            if i < 0:
                self.BadSyntax(argstr, i, "EOF found when expecting '{'")
        else:
            i = j

        if argstr[i : i + 1] != "{":
            return -1  # the node wasn't part of a graph

        j = i + 1

        if self._context is not None:
            self.BadSyntax(argstr, i, "Nested graphs are not allowed")

        oldParentContext = self._parentContext  # noqa: N806
        self._parentContext = self._context
        reason2 = self._reason2
        self._reason2 = becauseSubGraph
        # type error: Incompatible types in assignment (expression has type "Graph", variable has type "Optional[Formula]")
        self._context = self._store.newGraph(graph)  # type: ignore[assignment]

        while 1:
            i = self.skipSpace(argstr, j)
            if i < 0:
                self.BadSyntax(argstr, i, "needed '}', found end.")

            if argstr[i : i + 1] == "}":
                j = i + 1
                break

            j = self.directiveOrStatement(argstr, i)
            if j < 0:
                self.BadSyntax(argstr, i, "expected statement or '}'")

        self._context = self._parentContext
        self._reason2 = reason2
        self._parentContext = oldParentContext
        # res.append(subj.close())    # No use until closed
        return j

### `` labelOrSubject

    labelOrSubject(argstr: str, i: int, res: MutableSequence[Any]) -> int

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

    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60

    def labelOrSubject(  # noqa: N802
        self, argstr: str, i: int, res: MutableSequence[Any]
    ) -> int:
        j = self.skipSpace(argstr, i)
        if j < 0:
            return j  # eof
        i = j

        j = self.uri_ref2(argstr, i, res)
        if j >= 0:
            return j

        if argstr[i] == "[":
            j = self.skipSpace(argstr, i + 1)
            if j < 0:
                self.BadSyntax(argstr, i, "Expected ] got EOF")
            if argstr[j] == "]":
                res.append(self.blankNode())
                return j + 1
        return -1

## `` becauseSubGraph

    becauseSubGraph(*args, **kwargs)

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

    11
    12

    def becauseSubGraph(*args, **kwargs):  # noqa: N802
        pass