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

typing and cleanup

Browse source
This commit is contained in:
boyska 2022-01-17 00:21:13 +01:00
parent dcbb804b94
commit d9d0af994c
1 changed files with 8 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
larigira/db.py
View file

@ -3,6 +3,8 @@ from tinydb import TinyDB
from tinydb.storages import JSONStorage from tinydb.storages import JSONStorage
from tinydb.middlewares import Middleware from tinydb.middlewares import Middleware
from pathlib import Path from pathlib import Path
from typing import Union, Tuple
class ReadOnlyMiddleware(Middleware): class ReadOnlyMiddleware(Middleware):
""" """
@ -20,6 +22,7 @@ class ReadOnlyMiddleware(Middleware):
class ReadOnlyException(ValueError): class ReadOnlyException(ValueError):
pass pass
class EventModel(object): class EventModel(object):
def __init__(self, uri, additional_db_dir=None): def __init__(self, uri, additional_db_dir=None):
self.uri = uri self.uri = uri
@ -60,14 +63,14 @@ 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): def canonicalize(self, 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): def parse_id(self, eid_or_aid: Union[str, int]) -> Tuple[str, int]:
try: try:
int(eid_or_aid) int(eid_or_aid)
except ValueError: except ValueError:
@ -79,9 +82,9 @@ class EventModel(object):
return (dbname, int(num)) return (dbname, int(num))
def get_action_by_id(self, action_id): def get_action_by_id(self, action_id: Union[str, int]):
db, action_id = self.parse_id(action_id) db, db_action_id = self.parse_id(action_id)
return self._dbs[db].table('actions').get(eid=action_id) return self._dbs[db].table('actions').get(eid=db_action_id)
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.parse_id(alarm_id)