# Copyright (C) 2018-2019 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 os import urllib.parse import yaml import ick2 class PersistentStateInterface: # pragma: no cover def get_resource_names(self, token, kind): raise NotImplementedError() def has_resource(self, token, kind, name): raise NotImplementedError() def get_resource(self, token, kind, name): raise NotImplementedError() def get_resources(self, token, kind): return [ self.get_resource(token, kind, name) for name in self.get_resource_names(token, kind) ] def write_resource(self, token, kind, name, resource): raise NotImplementedError() def update_resource(self, token, kind, name, resource): raise NotImplementedError() def remove_resource(self, token, kind, name): raise NotImplementedError() class MemoryPersistentState(PersistentStateInterface): def __init__(self): self._res = {} def get_resource_names(self, token, kind): if kind not in self._res: return [] return list(self._res[kind].keys()) def has_resource(self, token, kind, name): return kind in self._res and name in self._res[kind] def get_resource(self, token, kind, name): if kind not in self._res or name not in self._res[kind]: raise ick2.NotFound(kind=kind, name=name) return self._res[kind][name] def write_resource(self, token, kind, name, resource): if kind not in self._res: self._res[kind] = {} self._res[kind][name] = resource def update_resource(self, token, kind, name, resource): self.write_resource(token, kind, name, resource) def remove_resource(self, token, kind, name): if kind not in self._res or name not in self._res[kind]: raise ick2.NotFound(kind=kind, name=name) del self._res[kind][name] class NotFound(Exception): def __init__(self, kind, name): super().__init__( 'Resource {}:{} not found'.format( kind or "unknown", name or "unknown")) class Resource: # pragma: no cover def __init__(self, as_dict=None): self._dict = copy.deepcopy(as_dict or {}) def as_dict(self): return copy.deepcopy(self._dict) def __getitem__(self, key): return self._dict[key] def __setitem__(self, key, value): self._dict[key] = value def __contains__(self, key): return key in self._dict def __len__(self): return len(self._dict) def get(self, key, default=None): return self._dict.get(key, default) def from_dict(self, as_dict): self._dict.clear() for key in as_dict: self[key] = as_dict[key] def resource_from_dict(as_dict): # pragma: no cover return Resource(as_dict)