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` –

  Parses SPARQL CSV results into a Result object.

- `CSVResultSerializer` –

  Serializes SPARQL results into CSV format.

## `` CSVResultParser

    CSVResultParser()

Bases: `ResultParser`

Parses SPARQL CSV results into a Result object.

Methods:

- `convertTerm` –
- `parse` –
- `parseRow` –

Attributes:

- `delim` –

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:

- `serialize` –
- `serializeTerm` –

Attributes:

- `delim` –

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