boyska/larigira
1
0
Fork 0
Code Issues 13 Pull requests Releases Activity

change operations only work on "main" DB

Browse source
This commit is contained in:
boyska 2022-01-17 00:46:24 +01:00
parent 257e3e45cd
commit eb0f6c0310
1 changed files with 25 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
larigira/db.py
View file

@ -22,6 +22,16 @@ class ReadOnlyMiddleware(Middleware):
class ReadOnlyException(ValueError): class ReadOnlyException(ValueError):
pass pass
def only_main(f):
'''assumes first argument is id, and must be "main"'''
def wrapper(self, *args):
_id = args[0]
db, db_id = EventModel.parse_id(_id)
if db != 'main':
raise ReadOnlyException('You called a write operation on a readonly db')
return f(self, db_id, *args[1:])
return wrapper
class EventModel(object): class EventModel(object):
def __init__(self, uri, additional_db_dir=None): def __init__(self, uri, additional_db_dir=None):
@ -63,31 +73,36 @@ class EventModel(object):
self._actions = self._dbs['main'].table("actions") self._actions = self._dbs['main'].table("actions")
self._alarms = self._dbs['main'].table("alarms") self._alarms = self._dbs['main'].table("alarms")
def canonicalize(self, eid_or_aid: Union[str, int]) -> str: @staticmethod
def canonicalize(eid_or_aid: Union[str, int]) -> str:
try: try:
int(eid_or_aid) int(eid_or_aid)
except ValueError: except ValueError:
return eid_or_aid return eid_or_aid
return 'main:%d' % eid_or_aid return 'main:%d' % eid_or_aid
def parse_id(self, eid_or_aid: Union[str, int]) -> Tuple[str, int]: @staticmethod
def parse_id(eid_or_aid: Union[str, int]) -> Tuple[str, int]:
try: try:
int(eid_or_aid) int(eid_or_aid)
except ValueError: except ValueError:
pass pass
else: else:
return ('main', eid_or_aid) return ('main', int(eid_or_aid))
dbname, num = eid_or_aid.split(':') dbname, num = eid_or_aid.split(':')
return (dbname, int(num)) return (dbname, int(num))
def get_action_by_id(self, action_id: Union[str, int]): def get_action_by_id(self, action_id: Union[str, int]):
db, db_action_id = self.parse_id(action_id) canonical = self.canonicalize(action_id)
return self._dbs[db].table('actions').get(eid=db_action_id) db, db_action_id = self.__class__.parse_id(canonical)
out = self._dbs[db].table('actions').get(eid=db_action_id)
out.doc_id = canonical
return out
def get_alarm_by_id(self, alarm_id): def get_alarm_by_id(self, alarm_id):
db, alarm_id = self.parse_id(alarm_id) db, alarm_id = self.__class__.parse_id(alarm_id)
return self._dbs[db].table('alarms').get(eid=alarm_id) return self._dbs[db].table('alarms').get(eid=alarm_id)
def get_actions_by_alarm(self, alarm): def get_actions_by_alarm(self, alarm):
@ -125,14 +140,18 @@ class EventModel(object):
def add_alarm(self, alarm): def add_alarm(self, alarm):
return self.add_event(alarm, []) return self.add_event(alarm, [])
@only_main
def update_alarm(self, alarmid, new_fields={}): def update_alarm(self, alarmid, new_fields={}):
return self._alarms.update(new_fields, eids=[alarmid]) return self._alarms.update(new_fields, eids=[alarmid])
@only_main
def update_action(self, actionid, new_fields={}): def update_action(self, actionid, new_fields={}):
return self._actions.update(new_fields, eids=[actionid]) return self._actions.update(new_fields, eids=[actionid])
@only_main
def delete_alarm(self, alarmid): def delete_alarm(self, alarmid):
return self._alarms.remove(eids=[alarmid]) return self._alarms.remove(eids=[alarmid])
@only_main
def delete_action(self, actionid): def delete_action(self, actionid):
return self._actions.remove(eids=[actionid]) return self._actions.remove(eids=[actionid])