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
|
##
# .types.namedtuple - return rows as namedtuples
##
"""
Factories for namedtuple row representation.
"""
from collections import namedtuple
#: Global namedtuple type cache.
cache = {}
# Build and cache the namedtuple's produced.
def _factory(colnames : [str], namedtuple = namedtuple) -> tuple:
global cache
# Provide some normalization.
# Anything beyond this can just get renamed.
colnames = tuple([
x.replace(' ', '_') for x in colnames
])
try:
return cache[colnames]
except KeyError:
NT = namedtuple('row', colnames, rename = True)
cache[colnames] = NT
return NT
def NamedTupleFactory(attribute_map, composite_relid = None):
"""
Alternative db.typio.RowFactory for producing namedtuple's instead of
postgresql.types.Row() instances.
To install::
>>> from postgresql.types.namedtuple import NamedTupleFactory
>>> import postgresql
>>> db = postgresql.open(...)
>>> db.typio.RowTypeFactory(NamedTupleFactory)
And **all** Rows produced by that connection will be namedtuple()'s.
This includes composites.
"""
colnames = list(attribute_map.items())
colnames.sort(key = lambda x: x[1])
return lambda y: _factory((x[0] for x in colnames))(*y)
from itertools import chain, starmap
def namedtuples(stmt, from_iter = chain.from_iterable, map = starmap):
"""
Alternative to the .rows() execution method.
Use::
>>> from postgresql.types.namedtuple import namedtuples
>>> ps = namedtuples(db.prepare(...))
>>> for nt in ps(...):
... nt.a_column_name
This effectively selects the execution method to be used with the statement.
"""
NT = _factory(stmt.column_names)
# build the execution "method"
chunks = stmt.chunks
def rows_as_namedtuples(*args, **kw):
return map(NT, from_iter(chunks(*args, **kw))) # starmap
return rows_as_namedtuples
del chain, starmap
|
