Skip to content

tsvresults

This implements the Tab Separated SPARQL Result Format

It is implemented with pyparsing, reusing the elements from the SPARQL Parser

Classes:

Attributes:

EMPTY module-attribute 

EMPTY = FollowedBy(LineEnd()) | FollowedBy('\t')

HEADER module-attribute 

HEADER = Var + ZeroOrMore(Suppress('\t') + Var)

NONE_VALUE module-attribute 

NONE_VALUE = object()

RDFLITERAL module-attribute 

RDFLITERAL = Comp('literal', Param('string', String) + Optional(Param('lang', leave_whitespace()) | leave_whitespace() + leave_whitespace()))

ROW module-attribute 

ROW = (EMPTY | TERM) + ZeroOrMore(Suppress('\t') + (EMPTY | TERM))

String module-attribute 

String = STRING_LITERAL1 | STRING_LITERAL2

TERM module-attribute 

TERM = RDFLITERAL | IRIREF | BLANK_NODE_LABEL | NumericLiteral | BooleanLiteral

TSVResultParser 

TSVResultParser()

Bases: ResultParser

Parses SPARQL TSV results into a Result object.

Methods:

Source code in rdflib/query.py 
416
417
def __init__(self):
    pass

convertTerm 

convertTerm(t: Union[object, Literal, BNode, CompValue, URIRef]) -> Optional[Identifier]
Source code in rdflib/plugins/sparql/results/tsvresults.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def convertTerm(
    self, t: Union[object, RDFLiteral, BNode, CompValue, URIRef]
) -> typing.Optional[Identifier]:
    if t is NONE_VALUE:
        return None
    if isinstance(t, CompValue):
        if t.name == "literal":
            return RDFLiteral(t.string, lang=t.lang, datatype=t.datatype)
        else:
            raise Exception("I dont know how to handle this: %s" % (t,))
    elif isinstance(t, (RDFLiteral, BNode, URIRef)):
        return t
    else:
        raise ValueError(f"Unexpected type {type(t)} found in TSV result")

parse 

parse(source: IO, content_type: Optional[str] = None) -> Result
Source code in rdflib/plugins/sparql/results/tsvresults.py 
 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
def parse(self, source: IO, content_type: typing.Optional[str] = None) -> Result:  # type: ignore[override]
    if isinstance(source.read(0), bytes):
        # if reading from source returns bytes do utf-8 decoding
        # type error: Incompatible types in assignment (expression has type "StreamReader", variable has type "IO[Any]")
        source = codecs.getreader("utf-8")(source)  # type: ignore[assignment]

    r = Result("SELECT")

    header = source.readline()

    r.vars = list(HEADER.parse_string(header.strip(), parse_all=True))
    r.bindings = []
    while True:
        line = source.readline()
        if not line:
            break
        line = line.strip("\n")
        if line == "":
            continue

        row = ROW.parse_string(line, parse_all=True)
        this_row_dict = {}
        for var, val_read in zip(r.vars, row):
            val = self.convertTerm(val_read)
            if val is None:
                # Skip unbound vars
                continue
            this_row_dict[var] = val
        # Preserve solution row cardinality, including fully-unbound rows.
        r.bindings.append(this_row_dict)
    return r