File: extras.py

package info (click to toggle)
python-clickhouse-driver 0.2.5-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 2,516 kB
  • sloc: python: 10,950; pascal: 42; makefile: 31; sh: 3
file content (73 lines) | stat: -rw-r--r-- 2,129 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import re
from collections import namedtuple
from functools import lru_cache

from .cursor import Cursor


class DictCursor(Cursor):
    """
    A cursor that generates results as :class:`dict`.

    ``fetch*()`` methods will return dicts instead of tuples.
    """

    def fetchone(self):
        rv = super(DictCursor, self).fetchone()
        if rv is not None:
            rv = dict(zip(self._columns, rv))
        return rv

    def fetchmany(self, size=None):
        rv = super(DictCursor, self).fetchmany(size=size)
        return [dict(zip(self._columns, x)) for x in rv]

    def fetchall(self):
        rv = super(DictCursor, self).fetchall()
        return [dict(zip(self._columns, x)) for x in rv]


class NamedTupleCursor(Cursor):
    """
    A cursor that generates results as named tuples created by
    :func:`~collections.namedtuple`.

    ``fetch*()`` methods will return named tuples instead of regular tuples, so
    their elements can be accessed both as regular numeric items as well as
    attributes.
    """

    # ascii except alnum and underscore
    _re_clean = re.compile(
        '[' + re.escape(' !"#$%&\'()*+,-./:;<=>?@[\\]^`{|}~') + ']')

    @classmethod
    @lru_cache(512)
    def _make_nt(self, key):
        fields = []
        for s in key:
            s = self._re_clean.sub('_', s)
            # Python identifier cannot start with numbers, namedtuple fields
            # cannot start with underscore.
            if s[0] == '_' or '0' <= s[0] <= '9':
                s = 'f' + s
            fields.append(s)

        return namedtuple('Record', fields)

    def fetchone(self):
        rv = super(NamedTupleCursor, self).fetchone()
        if rv is not None:
            nt = self._make_nt(self._columns)
            rv = nt(*rv)
        return rv

    def fetchmany(self, size=None):
        rv = super(NamedTupleCursor, self).fetchmany(size=size)
        nt = self._make_nt(self._columns)
        return [nt(*x) for x in rv]

    def fetchall(self):
        rv = super(NamedTupleCursor, self).fetchall()
        nt = self._make_nt(self._columns)
        return [nt(*x) for x in rv]