Simplify db/model handling

Author: ZtF

bin/indoktrinator

@@ -14,10 +14,8 @@ from twisted.web.server import Site
 from twisted.python.threadpool import ThreadPool
 from twisted.python import log
 
-# Data are accessed through SQLSoup, using SQLAlchemy.
-from sqlalchemy.orm import scoped_session, sessionmaker
 from sqlalchemy import create_engine
-from indoktrinator.sqlalchemy_db import SQLSoup
+from indoktrinator.sqlalchemy_db import SQLAlchemyDB
 
 # Command line arguments follow the GNU conventions.
 from getopt import gnu_getopt
@@ -65,9 +63,7 @@ if __name__ == '__main__':
         notifier = Notifier(reactor, db_url)
 
         # Prepare database connection with table reflection.
-        engine = create_engine(db_url)
-        session = scoped_session(sessionmaker())
-        db = SQLSoup(engine, session=session)
+        db = SQLAlchemyDB(create_engine(db_url))
 
         # Prepare a 0MQ router instance for communication with the
         # leader that publishes our indoctrination schedule.
@@ -139,14 +135,10 @@ if __name__ == '__main__':
             config_path = v
 
     # Load the configuration from file.
+    config = ConfigParser()
     if action not in (do_help, do_version):
-        config = ConfigParser()
         config.read(config_path)
 
-    # Load the configuration from file.
-    config = ConfigParser()
-    config.read(config_path)
-
     # Perform the selected action.
     action(config=config)
 

bin/indoktrinator-harvester

@@ -6,10 +6,8 @@ from twisted.internet import reactor
 from twisted.web.wsgi import WSGIResource
 from twisted.python import log
 
-# Data are accessed through SQLSoup, using SQLAlchemy.
-from sqlalchemy.orm import scoped_session, sessionmaker
 from sqlalchemy import create_engine
-from indoktrinator.sqlalchemy_db import SQLSoup
+from indoktrinator.sqlalchemy_db import SQLAlchemyDB
 
 # Command line arguments follow the GNU conventions.
 from getopt import gnu_getopt
@@ -34,9 +32,7 @@ if __name__ == '__main__':
         path = config.get('media', 'path', fallback='/var/lib/indoktrinator')
 
         # Prepare database connection with table reflection.
-        engine = create_engine(db_url)
-        session = scoped_session(sessionmaker(autoflush=True))
-        db = SQLSoup(engine, session=session)
+        db = SQLAlchemyDB(create_engine(db_url))
 
         # Extract harvester options, sans the pool_size we handle here.
         harvester_opts = dict(config.items('harvester'))

indoktrinator/db.py

@@ -5,7 +5,7 @@
 
 from psycopg2 import STRING
 from psycopg2.extras import RangeCaster
-from psycopg2.extensions import AsIs, register_type, new_type
+from psycopg2.extensions import register_type, new_type
 from sqlalchemy.types import UserDefinedType
 from sqlalchemy.dialects.postgresql.base import ischema_names
 

indoktrinator/harvester/__init__.py

@@ -200,6 +200,7 @@ def probe_done(info):
 
         if d is not None:
             d.addCallback(probe_done)
+            d.addErrback(lambda f: log.err(f, 'Failed to process {!r}'.format(path)))
 
     @with_session
     def update_item_with_info(self, playlist, item, node, info):

indoktrinator/harvester/shadow_tree.py

@@ -99,8 +99,7 @@ def scan(self):
                     self.add_child(fname, False)
 
             for child in missing:
-                child = self.children.pop(missing)
-                child.lost()
+                self.children.pop(child).lost()
 
             break
 

indoktrinator/manager/__init__.py

@@ -10,7 +10,6 @@
 from uuid import uuid4
 from time import time
 
-from indoktrinator.db import with_session
 from indoktrinator.manager.schema import schema
 from indoktrinator.manager.scheduler import *
 from indoktrinator.manager.store import *
@@ -37,6 +36,7 @@ def __init__(self, db, notifier, router, media_path, url,
         self.power_down_gap = int(power_down_gap)
 
         # Collection of devices we have come into contact with.
+        # Written only from the reactor thread; HTTP threads read via dict.copy().
         self.devices = {}
 
         # Store for database objects received via notifications.
@@ -158,8 +158,9 @@ def on_program_change(self, uuid, program):
         else:
             log.msg('Program {} deleted.'.format(uuid))
 
-            if uuid in self.plans:
-                self.plans.pop(uuid)
+            str_uuid = str(uuid)
+            if str_uuid in self.plans:
+                del self.plans[str_uuid]
 
         self.sync_devices()
 
@@ -178,14 +179,18 @@ def update_plans(self):
 
     def on_device_change(self, id, device):
         """Device has changed in the database."""
-        self.sync_device(id)
+        if device is None:
+            # Device deleted — remove runtime state.
+            self.devices.pop(id, None)
+        else:
+            self.sync_device(id)
 
     def sync_devices(self):
         """
         Synchronize all devices plans as needed.
         """
 
-        for id in self.devices:
+        for id in list(self.devices):
             self.sync_device(id)
 
     def sync_device(self, id):
@@ -211,7 +216,12 @@ def sync_device(self, id):
             return
 
         # Find plan for device program.
-        plan = self.plans[str(device['program'])]
+        plan = self.plans.get(str(device['program']))
+
+        if plan is None:
+            # Plan not computed yet (e.g. during startup or after DB change).
+            self.send(id, 'plan', EMPTY_PLAN)
+            return
 
         # Check that device plan is up to date.
         if status['plan'] != plan['id']:

indoktrinator/manager/scheduler.py

@@ -12,8 +12,6 @@
 from re import findall
 from uuid import uuid4
 
-from indoktrinator.db import with_session
-
 
 __all__ = ['make_plan', 'EMPTY_PLAN']
 

indoktrinator/site/__init__.py

@@ -191,7 +191,8 @@ def api_devices(depth, **kwargs):
                     )
                 )
 
-            for devid, status in manager.devices.items():
+            # Snapshot to avoid RuntimeError if reactor modifies devices concurrently.
+            for devid, status in manager.devices.copy().items():
                 if devid not in persistent:
                     devices.insert(
                         0,
@@ -218,8 +219,8 @@ def api_device(id, depth, **kwargs):
             try:
                 device = model.device.get(id, depth=depth)
             except KeyError:
-                device = manager.devices.get(id)
-                device.update({'pending': True})
+                device = dict(manager.devices.get(id) or {})
+                device['pending'] = True
 
             status = manager.devices.get(id, {})
 
@@ -237,13 +238,10 @@ def api_device(id, depth, **kwargs):
             try:
                 return jsonify(deleted=model.device.delete(id))
             except KeyError:
-                pass
-            finally:
                 if id not in manager.devices:
                     log.msg('No device with id: {id}'.format(id=id))
-                else:
-                    del manager.devices[id]
                 return jsonify(deleted=id)
+            # Reactor cleans up manager.devices via on_device_change when NOTIFY arrives.
 
         if 'PATCH' == request.method:
             patch = request.get_json(force=True)

indoktrinator/site/model.py

@@ -7,6 +7,10 @@
 
 __all__ = ['Model']
 
+# Maps SQLAlchemy mapped classes to their fixup function.
+# Populated by Table.__init__ so dbdict() needs no magic on the ORM objects.
+_fixup_registry: dict = {}
+
 
 class Table:
     NAME = None
@@ -23,15 +27,8 @@ def __init__(self, model, db):
         self.db = db
         self.table = getattr(db, self.NAME)
 
-        for key, other_table in self.RELS:
-            entity = getattr(db, other_table)
-            ordering = []
-
-            for column in self.model.TABLES[other_table].ORDER_BY:
-                ordering.append(getattr(entity, column))
-
-        # Fox the `dbdict` function.
-        self.table._table.fixup = self.fixup
+        # Register this table's fixup so dbdict() can find it by mapped class.
+        _fixup_registry[self.table._cls] = self.fixup
 
     def get(self, key, depth=0):
         obj = self.table.get(key)
@@ -243,7 +240,7 @@ def dbdict(obj, depth=0):
     Convert an SQLAlchemy object to a dictionary.
     """
 
-    fixup = obj._table.fixup
+    fixup = _fixup_registry.get(type(obj), lambda x: x)
     mapper = class_mapper(obj.__class__)
     data = {col.key: getattr(obj, col.key) for col in mapper.columns}