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
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
"""Monadic utilities
This module is a collection of monadic utilities for use in
GeneNetwork. It includes:
* MonadicDict - monadic version of the built-in dictionary
* MonadicDictCursor - monadic version of MySQLdb.cursors.DictCursor
that returns a MonadicDict instead of the built-in dictionary
"""
from collections import UserDict
from functools import partial
from typing import Any, Hashable, Iterator
from MySQLdb.cursors import DictCursor, SSDictCursor
from pymonad.maybe import Maybe, Just, Nothing
class MonadicDict(UserDict):
"""
Monadic version of the built-in dictionary.
Keys in this dictionary can be any python object, but values must
be monadic values.
from pymonad.maybe import Just, Nothing
Initialize by setting individual keys to monadic values.
>>> d = MonadicDict()
>>> d["foo"] = Just(1)
>>> d["bar"] = Nothing
>>> d
{'foo': 1}
Initialize by converting a built-in dictionary object.
>>> MonadicDict({"foo": 1})
{'foo': 1}
>>> MonadicDict({"foo": 1, "bar": None})
{'foo': 1}
Initialize from a built-in dictionary object with monadic values.
>>> MonadicDict({"foo": Just(1)}, convert=False)
{'foo': 1}
>>> MonadicDict({"foo": Just(1), "bar": Nothing}, convert=False)
{'foo': 1}
Get values. For non-existent keys, Nothing is returned. Else, a
Just value is returned.
>>> d["foo"]
Just 1
>>> d["bar"]
Nothing
Convert MonadicDict object to a built-in dictionary object.
>>> d.data
{'foo': 1}
>>> type(d)
<class 'utility.monads.MonadicDict'>
>>> type(d.data)
<class 'dict'>
Delete keys. Deleting non-existent keys does nothing.
>>> del d["bar"]
>>> d
{'foo': 1}
>>> del d["foo"]
>>> d
{}
Update dictionary object.
>>> d = MonadicDict()
>>> d.update(MonadicDict({'foo': 'bar'}))
>>> d
{'foo': 'bar'}
"""
# pylint: disable=dangerous-default-value
def __init__(self, d: dict = {}, convert: bool = True) -> None:
"""Initialize monadic dictionary.
If convert is False, values in dictionary d must be
monadic. If convert is True, values in dictionary d are
converted to monadic values.
"""
if convert:
super().__init__(
{
key: Just(value)
for key, value in d.items()
if value is not None
}
)
else:
super().__init__(d)
def __getitem__(self, key: Hashable) -> Maybe[Any]:
"""Get key from dictionary.
If key exists in the dictionary, return a Just value. Else,
return Nothing.
"""
try:
return Just(self.data[key])
except KeyError:
return Nothing
def __setitem__(self, key: Hashable, value: Any) -> None:
"""Set key in dictionary.
value must be a monadic value---either Nothing or a Just
value. If value is a Just value, set it in the dictionary. If
value is Nothing, do nothing.
"""
value.bind(partial(super().__setitem__, key))
def __delitem__(self, key: Hashable) -> None:
"""Delete key from dictionary.
If key exists in the dictionary, delete it. Else, do nothing.
"""
try:
super().__delitem__(key)
except KeyError:
pass
def query_sql(conn, query: str, server_side: bool=False) -> Iterator[MonadicDict]:
"""Execute SQL query and return a generator of MonadicDict objects.
If server_side is False, the result set is stored in the client. If
server_side is True, the result set is stored in the server. Therefore,
when server_side is True, this query must be completed before a new one
can be executed.
"""
with conn.cursor(SSDictCursor if server_side else DictCursor) as cursor:
cursor.execute(query)
while row := cursor.fetchone():
yield MonadicDict(row)
|
