Skip to content

csvresults

This module implements a parser and serializer for the CSV SPARQL result formats

http://www.w3.org/TR/sparql11-results-csv-tsv/

Classes:

CSVResultParser 

CSVResultParser()

Bases: ResultParser

Parses SPARQL CSV results into a Result object.

Methods:

Attributes:

Source code in rdflib/plugins/sparql/results/csvresults.py 
23
24
def __init__(self):
    self.delim = ","

delim instance-attribute 

delim = ','

convertTerm 

convertTerm(t: str) -> Optional[Union[BNode, URIRef, Literal]]
Source code in rdflib/plugins/sparql/results/csvresults.py 
54
55
56
57
58
59
60
61
def convertTerm(self, t: str) -> Optional[Union[BNode, URIRef, Literal]]:
    if t == "":
        return None
    if t.startswith("_:"):
        return BNode(t)  # or generate new IDs?
    if t.startswith("http://") or t.startswith("https://"):  # TODO: more?
        return URIRef(t)
    return Literal(t)

parse 

parse(source: IO, content_type: Optional[str] = None) -> Result
Source code in rdflib/plugins/sparql/results/csvresults.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def parse(self, source: IO, content_type: Optional[str] = None) -> Result:  # type: ignore[override]
    r = Result("SELECT")

    # type error: Incompatible types in assignment (expression has type "StreamReader", variable has type "IO[Any]")
    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]

    reader = csv.reader(source, delimiter=self.delim)
    r.vars = [Variable(x) for x in next(reader)]
    r.bindings = []

    for row in reader:
        r.bindings.append(self.parseRow(row, r.vars))

    return r

parseRow 

parseRow(row: List[str], v: List[Variable]) -> Dict[Variable, Union[BNode, URIRef, Literal]]
Source code in rdflib/plugins/sparql/results/csvresults.py 
45
46
47
48
49
50
51
52
def parseRow(
    self, row: List[str], v: List[Variable]
) -> Dict[Variable, Union[BNode, URIRef, Literal]]:
    return dict(
        (var, val)
        for var, val in zip(v, [self.convertTerm(t) for t in row])
        if val is not None
    )

CSVResultSerializer 

CSVResultSerializer(result: SPARQLResult)

Bases: ResultSerializer

Serializes SPARQL results into CSV format.

Methods:

Attributes:

Source code in rdflib/plugins/sparql/results/csvresults.py 
67
68
69
70
71
72
def __init__(self, result: SPARQLResult):
    ResultSerializer.__init__(self, result)

    self.delim = ","
    if result.type != "SELECT":
        raise Exception("CSVSerializer can only serialize select query results")

delim instance-attribute 

delim = ','

serialize 

serialize(stream: IO, encoding: str = 'utf-8', **kwargs) -> None
Source code in rdflib/plugins/sparql/results/csvresults.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def serialize(self, stream: IO, encoding: str = "utf-8", **kwargs) -> None:
    # the serialiser writes bytes in the given encoding
    # in py3 csv.writer is unicode aware and writes STRINGS,
    # so we encode afterward

    import codecs

    # TODO: Find a better solution for all this casting
    writable_stream = cast(Union[TextIOBase, BufferedIOBase], stream)
    if isinstance(writable_stream, TextIOBase):
        string_stream: TextIOBase = writable_stream
    else:
        byte_stream = cast(BufferedIOBase, writable_stream)
        string_stream = cast(TextIOBase, codecs.getwriter(encoding)(byte_stream))

    out = csv.writer(string_stream, delimiter=self.delim)

    vs = [self.serializeTerm(v, encoding) for v in self.result.vars]  # type: ignore[union-attr]
    out.writerow(vs)
    for row in self.result.bindings:
        out.writerow(
            [self.serializeTerm(row.get(v), encoding) for v in self.result.vars]  # type: ignore[union-attr]
        )

serializeTerm 

serializeTerm(term: Optional[Identifier], encoding: str) -> Union[str, Identifier]
Source code in rdflib/plugins/sparql/results/csvresults.py 
 98
 99
100
101
102
103
104
105
106
def serializeTerm(
    self, term: Optional[Identifier], encoding: str
) -> Union[str, Identifier]:
    if term is None:
        return ""
    elif isinstance(term, BNode):
        return f"_:{term}"
    else:
        return term