# Copyright (C) 2018 Lars Wirzenius # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program 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 Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import copy import muck class MemoryStore: def __init__(self): self._dict = {} def __len__(self): return len(self._dict) def as_dict(self): return self._dict def __contains__(self, key): return key in self._dict def __getitem__(self, key): meta, res = self._dict[key] meta = copy.deepcopy(meta) res = copy.deepcopy(res) return meta, res def search(self, conds): hits = [] for rid in self._dict: meta, res = self._dict[rid] if any(self._matches(meta, res, cond) for cond in conds): hits.append(rid) return hits def _matches(self, meta, res, cond): if cond['where'] not in ('meta', 'data'): # pragma: no cover return False if cond['where'] == 'meta': thing = meta else: thing = res field = cond['field'] pattern = cond['pattern'] if field not in thing: # pragma: no cover return False value = thing[field] funcs = { '==': self._equal, '>=': self._ge, '<=': self._le, } if cond['op'] not in funcs: # pragma: no cover return False is_match = funcs[cond['op']] if isinstance(value, list): return any(is_match(pattern, item) for item in value) return is_match(pattern, value) def _equal(self, expected, actual): return expected.lower() == actual.lower() def _ge(self, expected, actual): return actual.lower() >= expected.lower() def _le(self, expected, actual): return actual.lower() <= expected.lower() def change(self, chg): funcs = { 'create': self._create, 'update': self._update, 'delete': self._delete, } op = chg.get_op() func = funcs[op] func(chg) def _create(self, chg): rid = chg.get_id() if rid in self._dict: raise muck.Error('Resource {} already exists'.format(rid)) self._dict[rid] = (chg.get_meta(), chg.get_res()) def _update(self, chg): rid = chg.get_id() if rid not in self._dict: raise muck.Error('Resource {} does not exist'.format(rid)) self._dict[rid] = (chg.get_meta(), chg.get_res()) def _delete(self, chg): rid = chg.get_id() if rid not in self._dict: raise muck.Error('Resource {} does not exist'.format(rid)) del self._dict[rid]