aboutsummaryrefslogtreecommitdiff
path: root/gn3/monads.py
blob: 964dce2ac58d9dc2385b34ef5d1fe7cef0eb54ae (about) (plain)
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
"""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
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) -> Iterator[MonadicDict]:
    """Execute SQL query and return a generator of MonadicDict objects."""
    with conn.cursor(DictCursor) as cursor:
        cursor.execute(query)
        while row := cursor.fetchone():
            yield MonadicDict(row)