Implements storage for per-connection data sc StorageController sc has

  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
"""Implements storage for per-connection data
>>> sc = StorageController()
>>> sc.has_storage('sid1')
False
>>> s1 = sc('sid1') # default storage engine
>>> sc.has_storage('sid1')
True
>>> sc.drop('sid1')
>>> sc.has_storage('sid1')
False
>>> s2 = sc('sid2', storage='DictStorage')
>>> s2.has_key('key1')
False
>>> s2['key1'] = 'value1'
>>> s2['key1']
'value1'
>>> s2.has_key('key1')
True
>>> s2_clone = sc('sid2')
>>> s2_clone['key1']
'value1'
>>> s2['bad value']
Traceback (most recent call last):
...
KeyError: 'bad value'
"""
import settings
class StorageError(Exception):
pass
class AbstractStorage(object):
"""A storage for per-connection data"""
sid = ''
def __init__(self, storage_id):
self.sid = storage_id
pass
def __getitem__(self, key):
if self.has_key(key):
return self.get(key)
else:
raise KeyError("No key '%s' in storage %s", (key, self))
def __setitem__(self, key, value):
self.set(key, value)
def get(self, key):
raise NotImplementedError()
def set(self, key, value):
raise NotImplementedError()
def delete(self, key):
raise NotImplementedError()
def has_key(self, key):
raise NotImplementedError()
def terminate(self):
pass
class DictStorage(dict, AbstractStorage):
def __init__(self, storage_id, *args, **kwargs):
AbstractStorage.__init__(self, storage_id, *args, **kwargs)
def get(key):
return self[key]
def set(self, key, value):
self[key] = value
def delete(self, key):
del self[key]
class StorageController(object):
"""manages storage instances """
store = {}
def __init__(self):
pass
@staticmethod
def find_storage(storage):
"""If given string - tries to expand it to storage class"""
if isinstance(storage, basestring):
try:
storage = globals()[storage]
except KeyError, e:
raise StorageError("Invalid default storage in settings: %s" %
e.message)
if isinstance(storage, type) and \
issubclass(storage, AbstractStorage):
return storage
else:
raise TypeError("Can't find storage %s", storage)
def has_storage(self, key):
return self.store.has_key(key)
def drop(self, key):
self.store[key].terminate()
del self.store[key]
def __call__(self, sid, storage=None):
try:
return self.store[sid]
except KeyError:
if not storage:
storage = settings.DEFAULT_STORAGE
storage = self.find_storage(storage)
new_storage = storage(sid)
self.store[sid] = new_storage
return new_storage
def main():
import doctest
doctest.testmod()
if __name__ == "__main__":
main()