1
0
Fork 0

refactor db code

Dieser Commit ist enthalten in:
boyska 2018-12-28 23:46:50 +01:00
Ursprung f027517c49
Commit d13474b239
4 geänderte Dateien mit 66 neuen und 65 gelöschten Zeilen

61
larigira/db.py Normale Datei
Datei anzeigen

@ -0,0 +1,61 @@
from tinydb import TinyDB
class EventModel(object):
def __init__(self, uri):
self.uri = uri
self.db = None
self.reload()
def reload(self):
if self.db is not None:
self.db.close()
self.db = TinyDB(self.uri, indent=2)
self._actions = self.db.table('actions')
self._alarms = self.db.table('alarms')
def get_action_by_id(self, action_id):
return self._actions.get(eid=action_id)
def get_alarm_by_id(self, alarm_id):
return self._alarms.get(eid=alarm_id)
def get_actions_by_alarm(self, alarm):
for action_id in alarm.get('actions', []):
action = self.get_action_by_id(action_id)
if action is None:
continue
yield action
def get_all_alarms(self):
return self._alarms.all()
def get_all_actions(self):
return self._actions.all()
def get_all_alarms_expanded(self):
for alarm in self.get_all_alarms():
for action in self.get_actions_by_alarm(alarm):
yield alarm, action
def add_event(self, alarm, actions):
action_ids = [self.add_action(a) for a in actions]
alarm['actions'] = action_ids
return self._alarms.insert(alarm)
def add_action(self, action):
return self._actions.insert(action)
def add_alarm(self, alarm):
return self.add_event(alarm, [])
def update_alarm(self, alarmid, new_fields={}):
return self._alarms.update(new_fields, eids=[alarmid])
def update_action(self, actionid, new_fields={}):
return self._actions.update(new_fields, eids=[actionid])
def delete_alarm(self, alarmid):
return self._alarms.remove(eids=[alarmid])
def delete_action(self, actionid):
return self._actions.remove(eids=[actionid])

Datei anzeigen

@ -2,77 +2,17 @@ from __future__ import print_function
from gevent import monkey
monkey.patch_all(subprocess=True)
import logging
logging.getLogger('mpd').setLevel(logging.WARNING)
from datetime import datetime, timedelta
import gevent
from gevent.queue import Queue
from tinydb import TinyDB
from .eventutils import ParentedLet, Timer
from .timegen import timegenerate
from .audiogen import audiogenerate
from .db import EventModel
class EventModel(object):
def __init__(self, uri):
self.uri = uri
self.db = None
self.reload()
def reload(self):
if self.db is not None:
self.db.close()
self.db = TinyDB(self.uri, indent=2)
self.actions = self.db.table('actions')
self.alarms = self.db.table('alarms')
def get_action_by_id(self, action_id):
return self.actions.get(eid=action_id)
def get_alarm_by_id(self, alarm_id):
return self.alarms.get(eid=alarm_id)
def get_actions_by_alarm(self, alarm):
for action_id in alarm.get('actions', []):
action = self.get_action_by_id(action_id)
if action is None: continue
yield action
def get_all_alarms(self):
return self.alarms.all()
def get_all_actions(self):
return self.actions.all()
def get_all_alarms_expanded(self):
for alarm in self.get_all_alarms():
for action in self.get_actions_by_alarm(alarm):
yield alarm, action
def add_event(self, alarm, actions):
action_ids = [self.add_action(a) for a in actions]
alarm['actions'] = action_ids
return self.alarms.insert(alarm)
def add_action(self, action):
return self.actions.insert(action)
def add_alarm(self, alarm):
return self.add_event(alarm, [])
def update_alarm(self, alarmid, new_fields={}):
return self.alarms.update(new_fields, eids=[alarmid])
def update_action(self, actionid, new_fields={}):
return self.actions.update(new_fields, eids=[actionid])
def delete_alarm(self, alarmid):
return self.alarms.remove(eids=[alarmid])
def delete_action(self, actionid):
return self.actions.remove(eids=[actionid])
logging.getLogger('mpd').setLevel(logging.WARNING)
class Monitor(ParentedLet):
'''

Datei anzeigen

@ -2,7 +2,7 @@ from __future__ import print_function
import argparse
import json
from .event import EventModel
from .db import EventModel
from .config import get_conf

Datei anzeigen

@ -6,7 +6,7 @@ monkey.patch_all(subprocess=True)
import pytest
from larigira.event import EventModel
from larigira.db import EventModel
@pytest.yield_fixture
@ -37,7 +37,7 @@ def test_add_multiple_alarms(db):
dict(kind='foo', a=3)])
assert len(db.get_all_alarms()) == 1
assert db.get_alarm_by_id(alarm_id) is not None
assert len(db.actions.all()) == 2
assert len(db.get_all_actions()) == 2
assert len(tuple(db.get_actions_by_alarm(
db.get_alarm_by_id(alarm_id)))) == 2