Merge pull request #303 from gdetrez/scanjobstore

Add scans and scan jobs to the database

docker-compose.yml

@@ -12,15 +12,15 @@ services:
     environment:
       - SOM_BACKEND_HOST=scanomatic-backend
       - PGHOST=db
-      - PGUSER=postgres
+      - PGUSER=scanomatic
       - PGDATABASE=scanomatic
 
   scanomatic-backend:
     image: phenomique/scanomatic:latest
     build: .
     environment:
       - PGHOST=db
-      - PGUSER=postgres
+      - PGUSER=scanomatic
       - PGDATABASE=scanomatic
     volumes:
       - scan-o-matic-home:/root/.scan-o-matic
@@ -29,8 +29,8 @@ services:
 
   db:
     image: postgres:10
-    environment:
-      - POSTGRES_DB=scanomatic
+    volumes:
+      - "./scripts/setupdb.sh:/docker-entrypoint-initdb.d/setupdb.sh:ro"
 
 volumes:
   scan-o-matic-home: {}

scanomatic/data/migrations/versions/39da65401ca3_create_scanjob_table.py

@@ -0,0 +1,58 @@
+"""create scanjob table
+
+Revision ID: 39da65401ca3
+Revises: 6736f58587af
+Create Date: 2018-02-26 11:29:24.432510
+
+"""
+from __future__ import absolute_import
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import ExcludeConstraint
+from sqlalchemy.sql import quoted_name
+
+
+# revision identifiers, used by Alembic.
+revision = '39da65401ca3'
+down_revision = '6736f58587af'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'scanjobs',
+        sa.Column('id', sa.Text(), primary_key=True),
+        sa.Column('name', sa.Text(), unique=True, nullable=False),
+        sa.Column('duration', sa.Interval(), nullable=False),
+        sa.Column('interval', sa.Interval(), nullable=False),
+        sa.Column(
+            'scanner_id',
+            sa.Text(),
+            sa.ForeignKey('scanners.id', name='fk_scanjob_scanner_id'),
+            nullable=False,
+        ),
+        sa.Column('start_time', sa.DateTime(timezone=True)),
+
+        # Create a constraint that prevents scanjobs to overlap for the same
+        # scanner id.  This a PostgreSQL specific construct.  quoted_name is
+        # used to pass a raw expression because it seems that the tsrange
+        # function is not available in sqlalchemy.
+        ExcludeConstraint(
+            ('scanner_id', '='),
+            (sa.Column(quoted_name(
+                '''
+                    tsrange(
+                        start_time AT TIME ZONE 'UTC',
+                        start_time AT TIME ZONE 'UTC' + duration,
+                        '[]'
+                    )
+                ''', quote=False)), '&&'),
+            where=(sa.Column('start_time').isnot(None)),
+            name='exclude_overlapping_scanjobs',
+        ),
+    )
+
+
+def downgrade():
+    pass

scanomatic/data/migrations/versions/da56e259f2e6_create_scans_table.py

@@ -0,0 +1,37 @@
+"""create scans table
+
+Revision ID: da56e259f2e6
+Revises: 39da65401ca3
+Create Date: 2018-02-26 18:14:31.166852
+
+"""
+from __future__ import absolute_import
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'da56e259f2e6'
+down_revision = '39da65401ca3'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'scans',
+        sa.Column('id', sa.Text(), primary_key=True),
+        sa.Column('start_time', sa.DateTime(timezone=True), nullable=False),
+        sa.Column('end_time', sa.DateTime(timezone=True), nullable=False),
+        sa.Column('digest', sa.Text(), nullable=False),
+        sa.Column(
+            'scanjob_id',
+            sa.Text(),
+            sa.ForeignKey('scanjobs.id', name='fk_scan_scanjob_id'),
+            nullable=False,
+        ),
+    )
+
+
+def downgrade():
+    pass

scanomatic/data/scanjobstore.py

@@ -0,0 +1,83 @@
+from __future__ import absolute_import
+from pytz import utc
+import sqlalchemy as sa
+from sqlalchemy.sql import and_
+from sqlalchemy.sql.expression import exists
+
+from scanomatic.models.scanjob import ScanJob
+
+
+class ScanJobStore(object):
+    class IntegrityError(Exception):
+        pass
+
+    def __init__(self, connection, dbmetadata):
+        self._connection = connection
+        self._table = dbmetadata.tables['scanjobs']
+
+    def add_scanjob(self, scanjob):
+        try:
+            self._connection.execute(self._table.insert().values(
+                duration=scanjob.duration,
+                id=scanjob.identifier,
+                interval=scanjob.interval,
+                name=scanjob.name,
+                scanner_id=scanjob.scanner_id,
+                start_time=scanjob.start_time,
+            ))
+        except sa.exc.IntegrityError as e:
+            raise self.IntegrityError(e)
+
+    def set_scanjob_start_time(self, id_, start_time):
+        try:
+            self._connection.execute(
+                self._table.update()
+                .where(self._table.c.id == id_)
+                .values(start_time=start_time)
+            )
+        except sa.exc.IntegrityError as e:
+            raise self.IntegrityError(e)
+
+    def get_scanjob_by_id(self, id_):
+        query = self._table.select().where(self._table.c.id == id_)
+        for scanjob in self._get_scanjobs(query):
+            return scanjob
+        else:
+            raise KeyError(id_)
+
+    def get_all_scanjobs(self):
+        query = self._table.select()
+        return self._get_scanjobs(query)
+
+    def _get_scanjobs(self, query):
+        for row in self._connection.execute(query):
+            if row['start_time'] is not None:
+                start_time = row['start_time'].astimezone(utc)
+            else:
+                start_time = None
+            yield ScanJob(
+                name=row['name'],
+                identifier=row['id'],
+                duration=row['duration'],
+                interval=row['interval'],
+                scanner_id=row['scanner_id'],
+                start_time=start_time,
+            )
+
+    def has_scanjob_with_name(self, name):
+        query = (
+            exists(self._table.select().where(self._table.c.name == name))
+            .select()
+        )
+        return self._connection.execute(query).scalar()
+
+    def get_current_scanjob_for_scanner(self, scanner_id, when):
+        query = self._table.select().where(
+            and_(
+                self._table.c.scanner_id == scanner_id,
+                self._table.c.start_time <= when,
+                self._table.c.start_time + self._table.c.duration >= when,
+            )
+        )
+        for scanjob in self._get_scanjobs(query):
+            return scanjob

scanomatic/data/scannerstore.py

@@ -2,37 +2,39 @@
 import sqlalchemy as sa
 from sqlalchemy.sql.expression import exists
 
-from .tables import scanners
 from scanomatic.models.scanner import Scanner
 
 
 class ScannerStore(object):
     class IntegrityError(Exception):
         pass
 
-    def __init__(self, connection):
+    def __init__(self, connection, dbmetadata):
         self._connection = connection
+        self._table = dbmetadata.tables['scanners']
 
     def add(self, scanner):
         try:
-            self._connection.execute(scanners.insert().values(
+            self._connection.execute(self._table.insert().values(
                 name=scanner.name, id=scanner.identifier,
             ))
         except sa.exc.IntegrityError as e:
             raise self.IntegrityError(e)
 
     def get_all(self):
-        query = scanners.select()
+        query = self._table.select()
         for row in self._connection.execute(query):
             yield Scanner(name=row['name'], identifier=row['id'])
 
     def get_scanner_by_id(self, id_):
-        query = scanners.select().where(scanners.c.id == id_)
+        query = self._table.select().where(self._table.c.id == id_)
         for row in self._connection.execute(query):
             return Scanner(name=row['name'], identifier=row['id'])
         else:
             raise KeyError(id)
 
     def has_scanner_with_id(self, id_):
-        query = exists(scanners.select().where(scanners.c.id == id_)).select()
+        query = exists(
+            self._table.select().where(self._table.c.id == id_)
+        ).select()
         return self._connection.execute(query).scalar()

scanomatic/data/scanstore.py

@@ -0,0 +1,49 @@
+from __future__ import absolute_import
+from pytz import utc
+import sqlalchemy as sa
+
+from scanomatic.models.scan import Scan
+
+
+class ScanStore(object):
+    class IntegrityError(Exception):
+        pass
+
+    def __init__(self, connection, dbmetadata):
+        self._connection = connection
+        self._table = dbmetadata.tables['scans']
+
+    def add_scan(self, scan):
+        try:
+            self._connection.execute(
+                self._table.insert().values(
+                    id=scan.identifier,
+                    start_time=scan.start_time,
+                    end_time=scan.end_time,
+                    digest=scan.digest,
+                    scanjob_id=scan.scanjob_id,
+                )
+            )
+        except sa.exc.IntegrityError as e:
+            raise self.IntegrityError(e)
+
+    def get_scan_by_id(self, id_):
+        query = self._table.select().where(self._table.c.id == id_)
+        for scan in self._get_scans(query):
+            return scan
+        else:
+            raise KeyError(id_)
+
+    def get_all_scans(self):
+        query = self._table.select()
+        return self._get_scans(query)
+
+    def _get_scans(self, query):
+        for row in self._connection.execute(query):
+            yield Scan(
+                id=row['id'],
+                start_time=row['start_time'].astimezone(utc),
+                end_time=row['end_time'].astimezone(utc),
+                digest=row['digest'],
+                scanjob_id=row['scanjob_id'],
+            )

scanomatic/io/scanning_store.py

@@ -1,10 +1,7 @@
 from __future__ import absolute_import
 from collections import defaultdict
-from datetime import datetime
-import pytz
 
 from scanomatic.models.scan import Scan
-from scanomatic.models.scanjob import ScanJob
 from scanomatic.models.scannerstatus import ScannerStatus
 
 
@@ -31,8 +28,6 @@ def __str__(self):
 class ScanningStore:
     def __init__(self):
         self._stores = {
-            Scan: {},
-            ScanJob: {},
             ScannerStatus: defaultdict(list),
         }
 
@@ -47,8 +42,6 @@ def add(self, item):
         store = self._get_store(klass)
         if item.identifier in store:
             raise DuplicateIdError(klass, item.identifier)
-        if klass is Scan and not self.exists(ScanJob, item.scanjob_id):
-            raise UnknownIdError(ScanJob, item.scanjob_id)
         store[item.identifier] = item
 
     def get(self, klass, id_):
@@ -78,14 +71,6 @@ def update(self, item):
             raise UnknownIdError(type(item), item.identifier)
         self._get_store(type(item))[item.identifier] = item
 
-    def get_current_scanjob(self, scanner_id, timepoint):
-        for job in self.find(ScanJob, scanner_id=scanner_id):
-            if job.is_active(timepoint):
-                return job
-
-    def has_current_scanjob(self, scanner_id, timepoint):
-        return self.get_current_scanjob(scanner_id, timepoint) is not None
-
     def get_scanner_status_list(self, scanner_id):
         return self._get_store(ScannerStatus)[scanner_id]