aboutsummaryrefslogtreecommitdiff
path: root/wqflask/utility/monads.py
blob: 3360fad223a46f79d938acb98101db8e386537a1 (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
"""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 MySQLdb.cursors import DictCursor
from pymonad.maybe import 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
    {}
    """
    def __init__(self, d={}, convert=True):
        """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):
        """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, value):
        """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):
        """Delete key from dictionary.

        If key exists in the dictionary, delete it. Else, do nothing.
        """
        try:
            super().__delitem__(key)
        except KeyError:
            pass

class MonadicDictCursor(DictCursor):
    """Monadic version of MySQLdb.cursors.DictCursor.

    Monadic version of MySQLdb.cursors.DictCursor that returns a
    MonadicDict instead of the built-in dictionary.

    Execute a SQL query and retrieve results as MonadicDict
    objects. Each row object in the following code is a MonadicDict.
    >>> with conn.cursor(MonadicDictCursor) as cursor:
    ...    cursor.execute("SELECT foo FROM bar")
    ...    for row in cursor.fetchall():
    ...        print(row)
    """
    def fetchone(self):
        return Just(MonadicDict(row)) if (row := super().fetchone()) else Nothing
    def fetchmany(self, size=None):
        return [MonadicDict(row) for row in super().fetchmany(size=size)]
    def fetchall(self):
        return [MonadicDict(row) for row in super().fetchall()]