File: store.py

package info (click to toggle)
python-pypump 0.7-3
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, bullseye, sid
  • size: 512 kB
  • sloc: python: 3,153; makefile: 152
file content (203 lines) | stat: -rw-r--r-- 6,428 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
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203

# This has been taken from "Waterworks"
# Commit ID: dc05a36ed34ab94b657bcadeb70ccc3187227b2d
# URL: https://github.com/Aeva/waterworks
#
# PyPump is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# PyPump is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with PyPump.  If not, see <http://www.gnu.org/licenses/>.

from __future__ import absolute_import

import json
import os
import re
import stat
import datetime

from pypump.exceptions import ValidationError, StoreException

# Regex taken from WTForms
EMAIL_REGEX = re.compile(r"^.+@[^.].*\.[a-z]{2,10}$", re.IGNORECASE)


def webfinger_validator(webfinger):
    """ Validates webfinger is correct - should look like user@host.tld """
    error = "Invalid webfinger. Should be in format username@host.tld"
    if not EMAIL_REGEX.match(webfinger):
        raise ValidationError(error)


class AbstractStore(dict):
    """
    This should act like a dictionary. This should be persistant and
    save upon setting a value. The interface to this object is::

    >>> store = AbstractStore.load()
    >>> store["my-key"] = "my-value"
    >>> store["my-key"]
    'my-value'

    This must save when "my-value" was set (in __setitem__). There
    should also be a .save method which should take the entire object
    and write them out.
    """

    prefix = None

    def __init__(self, *args, **kwargs):
        self.__validators = {}
        return super(AbstractStore, self).__init__(*args, **kwargs)

    def __prefix_key(self, key):
        """ This will add the prefix to the key if one exists on the store """
        # If there isn't a prefix don't bother
        if self.prefix is None:
            return key

        # Don't prefix key if it already has it
        if key.startswith(self.prefix + "-"):
            return key

        return "{0}-{1}".format(self.prefix, key)

    def __setitem__(self, key, *args, **kwargs):
        if key in self.__validators.keys():
            self.__validators[key](*args, **kwargs)

        key = self.__prefix_key(key)
        super(AbstractStore, self).__setitem__(key, *args, **kwargs)
        self.save()

    def __getitem__(self, key, *args, **kwargs):
        key = self.__prefix_key(key)
        return super(AbstractStore, self).__getitem__(key, *args, **kwargs)

    def __contains__(self, key, *args, **kwargs):
        key = self.__prefix_key(key)
        return super(AbstractStore, self).__contains__(key, *args, **kwargs)

    def set_validator(self, key, validator):
        self.__validators[key] = validator

    def save(self):
        """ Save all attributes in store """
        raise NotImplementedError("This is a dummy class, abstract")

    def export(self):
        """ Exports as dictionary """
        data = {}
        for key, value in self.items():
            data[key] = value

        return data

    @classmethod
    def load(cls, webfinger, pypump):
        """ This create and populate a store object """
        raise NotImplementedError("This is a dummy class, abstract")

    def __str__(self):
        return str(self.export())


class DummyStore(AbstractStore):
    """
    This doesn't persistantly store any data it just acts like
    a regular dictionary. This shouldn't be used for anything but
    testing as nothing will be stored on disk.
    """

    def save(self):
        pass

    @classmethod
    def load(cls, webfinger, pypump):
        return cls()


class JSONStore(AbstractStore):
    """
    Persistant dictionary-like storage

    Will write out all changes to disk as they're made
    NB: Will overwrite any changes made to disk not on class.
    """

    def __init__(self, data=None, filename=None, *args, **kwargs):
        if filename is None:
            filename = self.get_filename()
        self.filename = filename

        if data is None:
            data = {}

        super(JSONStore, self).__init__(data, *args, **kwargs)

    def update(self, *args, **kwargs):
        return_value = super(JSONStore, self).update(*args, **kwargs)
        self.save()
        return return_value

    def save(self):
        """ Saves dictionary to disk in JSON format. """
        if self.filename is None:
            raise StoreException("Filename must be set to write store to disk")

        # We need an atomic way of re-writing the settings, we also need to
        # prevent only overwriting part of the settings file (see bug #116).
        # Create a temp file and only then re-name it to the config
        filename = "{filename}.{date}.tmp".format(
            filename=self.filename,
            date=datetime.datetime.utcnow().strftime('%Y-%m-%dT%H_%M_%S.%f')
        )

        # The `open` built-in doesn't allow us to set the mode
        mode = stat.S_IRUSR | stat.S_IWUSR  # 0600
        fd = os.open(filename, os.O_WRONLY | os.O_CREAT, mode)
        fout = os.fdopen(fd, "w")
        fout.write(json.dumps(self.export()))
        fout.close()

        # Now we should remove the old config
        if os.path.isfile(self.filename):
            os.remove(self.filename)

        # Now rename the temp file to the real config file
        os.rename(filename, self.filename)

    @classmethod
    def get_filename(cls):
        """ Gets filename of store on disk """
        config_home = os.environ.get("XDG_CONFIG_HOME", "~/.config")
        config_home = os.path.expanduser(config_home)

        base_path = os.path.join(config_home, "PyPump")
        if not os.path.isdir(base_path):
            os.makedirs(base_path)

        return os.path.join(base_path, "credentials.json")

    @classmethod
    def load(cls, webfinger, pypump):
        """ Load JSON from disk into store object """
        filename = cls.get_filename()

        if os.path.isfile(filename):
            data = open(filename).read()
            data = json.loads(data)
            store = cls(data, filename=filename)
        else:
            store = cls(filename=filename)

        store.prefix = webfinger
        return store