Continuing initial work of building/tweaking models and the overall framework. Began writing tests for process tracker and fixing bugs.

Author: OpenDataAlex

Pipfile

@@ -10,6 +10,8 @@ sqlalchemy = "*"
 sqlalchemy-utils = "*"
 dateutil = "*"
 python-dateutil = "*"
+psycopg2 = "*"
+psycopg2-binary = "*"
 
 [requires]
 python_version = "3.7"

models/actor.py

@@ -0,0 +1,14 @@
+# SQLAlchemy Models
+# Models for Actor entities
+
+from sqlalchemy import Column, Integer, Sequence, String
+from sqlalchemy_utils.types import uuid
+from models.model_base import Base
+
+
+class Actor(Base):
+
+    __tablename__ = "actor_lkup"
+
+    actor_uuid = Column(uuid.UUIDType, primary_key=True)
+    actor_name = Column(String(250), nullable=False, unique=True)

models/extract.py

@@ -0,0 +1,28 @@
+# SQLAlchemy Models
+# Models for Extract (Data) entities
+
+from sqlalchemy import Column, ForeignKey, Integer, Sequence, String
+from sqlalchemy.orm import relationship
+
+from models.model_base import Base
+
+
+class Extract(Base):
+
+    __tablename__ = "extract_tracking"
+
+    extract_id = Column(Integer, Sequence('extract_tracking_extract_id_seq01'), primary_key=True)
+    extract_source_id = Column(Integer, ForeignKey("source_lkup.source_id"))
+    extract_filename = Column(String(750), nullable=False, unique=True)
+    extract_location_id = Column(Integer, ForeignKey('location_lkup.location_id'))
+
+
+class Location(Base):
+
+    __tablename__ = "location_lkup"
+
+    location_id = Column(Integer, Sequence('location_lkup_location_id_seq01'), primary_key=True)
+    location_name = Column(String(250), nullable=False, unique=True)
+    location_path = Column(String(250), nullable=False, unique=True)
+
+    extracts = relationship("Extract")

models/model_base.py

@@ -0,0 +1,9 @@
+# Base Model class for other data models.
+
+from dateutil import parser
+
+from sqlalchemy.ext.declarative import declarative_base
+
+Base = declarative_base()
+default_date = parser.parse('1900-01-01 00:00:00')
+

models/process.py

@@ -1,34 +1,142 @@
 # SQLAlchemy Models
 # Models for Process entities
 
-from dateutil import parser
 
-from sqlalchemy import Column, ForeignKey, DateTime, Integer, String
-from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy import Column, DateTime, ForeignKey, Integer, Sequence, String
 from sqlalchemy.orm import relationship
 from sqlalchemy_utils.types import uuid
 
-Base = declarative_base()
-default_date = parser.parse('1900-01-01 00:00:00')
+from models.model_base import default_date, Base
+
+
+class ErrorType(Base):
+
+    __tablename__ = 'error_type_lkup'
+
+    error_type_id = Column(Integer, Sequence('error_type_lkup_error_type_id_seq01'), primary_key=True)
+    error_type_name = Column(String(250), unique=True, nullable=False)
+
+    process_errors = relationship("ErrorTracking")
+
+    def __repr__(self):
+
+        return "<ErrorType (name=%s)>" % self.error_type_name
+
+
+class ErrorTracking(Base):
+
+    __tablename__ = 'error_tracking'
+
+    error_tracking_id = Column(Integer, Sequence('error_tracking_error_tracking_id_seq01'), primary_key=True)
+    error_type_id = Column(Integer, ForeignKey('error_type_lkup.error_type_id'))
+    error_description = Column(String(750))
+    error_occurrence_date_time = Column(DateTime, nullable=False)
+    process_tracking_id = Column(Integer, ForeignKey('process_tracking.process_tracking_id'))
+
+    error_tracking = relationship("ProcessTracking")
+
+    def __repr__(self):
+
+        return "<ErrorTracking (id=%s, type=%s, description=%s" \
+               ", occurrence_date=%s)>" % (self.error_tracking_id
+                                          , self.error_type_id
+                                          , self.error_description
+                                          , self.error_occurrence_date_time)
+
+
+class ProcessStatus(Base):
+
+    __tablename__ = 'process_status_lkup'
+
+    process_status_id = Column(Integer, primary_key=True)
+    process_status_name = Column(String(75), nullable=False, unique=True)
+
+    def __repr__(self):
+
+        return "<ProcessStatus (id=%s, process_status_name=%s)>" % (self.process_status_id, self.process_status_name)
 
 
 class ProcessType(Base):
-    __tablename__ = 'process_type'
 
+    __tablename__ = 'process_type_lkup'
+
+    process_type_id = Column(Integer, Sequence('process_type_lkup_process_type_id_seq01'), primary_key=True)
+    process_type_name = Column(String(250), nullable=False)
+
+    processes = relationship("Process")
+
+    def __repr__(self):
+
+        return "<ProcessType (id=%s, process_type=%s)>" % (self.process_type_id, self.process_type_name)
 
 
 class Process(Base):
+
     __tablename__ = 'process'
+
     process_uuid = Column(uuid.UUIDType(binary=False), primary_key=True)
     process_name = Column(String(250), nullable=False, unique=True)
-    latest_run_low_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
-    latest_run_high_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
-    latest_run_id = Column(Integer, nullable=False, default=0)
-    latest_run_start_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
-    latest_run_end_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
-    latest_run_process_status = Column(Integer, nullable=False, default=0)
-    latest_run_record_count = Column(Integer, nullable=False, default=0)
+    process_source_id = Column(Integer, ForeignKey('source_lkup.source_id'))
+#    latest_run_low_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
+#    latest_run_high_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
+#    latest_run_id = Column(Integer, nullable=False, default=0)
+#    latest_run_start_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
+#    latest_run_end_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
+#    latest_run_process_status = Column(Integer, nullable=False, default=0)
+#    latest_run_record_count = Column(Integer, nullable=False, default=0)
     total_record_count = Column(Integer, nullable=False, default=0)
-    latest_run_actor_id = Column()
-    process_type_id = Column()
+#    latest_run_actor_id = Column(uuid.UUIDType, ForeignKey('actor.actor_uuid'))
+    process_type_id = Column(Integer, ForeignKey('process_type_lkup.process_type_id'))
     last_failed_run_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
+
+    process_tracking = relationship("ProcessTracking")
+    process_type = relationship("ProcessType", back_populates="processes")
+    source = relationship("Source")
+
+    def __repr__(self):
+
+        return "<Process (uuid=%s, name=%s, source=%s, type=%s)>" % (self.process_uuid
+                                                                     , self.process_name
+                                                                     , self.process_source_id
+                                                                     , self.process_type_id)
+
+
+class ProcessDependency(Base):
+
+    __tablename__ = 'process_dependency'
+
+    parent_process_uuid = Column(uuid.UUIDType, ForeignKey('process.process_uuid'), primary_key=True)
+    child_process_uuid = Column(uuid.UUIDType, ForeignKey('process.process_uuid'), primary_key=True)
+
+    child_process = relationship('Process', foreign_keys=[child_process_uuid])
+    parent_process = relationship('Process', foreign_keys=[parent_process_uuid])
+
+    def __repr__(self):
+
+        return "<ProcessDependency (parent_process=%s, child_process=%s)>" % (self.parent_process_uuid
+                                                                              , self.child_process_uuid)
+
+
+class ProcessTracking(Base):
+
+    __tablename__ = 'process_tracking'
+
+    process_tracking_id = Column(Integer, Sequence('process_tracking_process_tracking_id_seq01'), primary_key=True)
+    process_uuid = Column(uuid.UUIDType, ForeignKey('process.process_uuid'))
+    process_status = Column(Integer, ForeignKey('process_status_lkup.process_type_id'))
+    process_run_id = Column(Integer, nullable=False)
+    process_run_low_date_time = Column(DateTime, nullable=True)
+    process_run_high_date_time = Column(DateTime, nullable=True)
+    process_run_start_date_time = Column(DateTime, nullable=False)
+    process_run_end_date_time = Column(DateTime, nullable=True)
+    process_run_record_count = Column(Integer, nullable=False, default=0)
+    process_run_actor_uuid = Column(uuid.UUIDType, ForeignKey('actor.actor_uuid'))
+
+    process = relationship("Process", back_populates="process_tracking")
+    errors = relationship("ErrorTracking", back_populates="error_tracking")
+
+    def __repr__(self):
+
+        return "<ProcessTracking id=%s, process=%s, process_status=%s)>" % (self.process_tracking_id
+                                                                            , self.process_uuid
+                                                                            , self.process_status)

models/source.py

@@ -0,0 +1,18 @@
+# SQLAlchemy Models
+# Models for Source entities
+
+from sqlalchemy import Column, Integer, Sequence, String
+
+from models.model_base import Base
+
+
+class Source(Base):
+
+    __tablename__ = "source_lkup"
+
+    source_id = Column(Integer, Sequence('source_lkup_source_id_seq01'), primary_key=True)
+    source_name = Column(String(250), nullable=False, unique=True)
+
+    def __repr__(self):
+
+        return "<Source (name=%s)>" % self.source_name

models/tool.py

@@ -0,0 +1,18 @@
+# SQLAlchemy Models
+# Models for Tool entities
+
+from sqlalchemy import Column, Integer, Sequence, String
+
+from models.model_base import Base
+
+
+class Tool(Base):
+
+    __tablename__ = "tool_lkup"
+
+    tool_id = Column(Integer, Sequence('tool_lkup_tool_id_seq01'), primary_key=True)
+    tool_name = Column(String(250), nullable=False, unique=True)
+
+    def __repr__(self):
+
+        return "<Tool (name=%s)>" % self.tool_name

process.py

@@ -0,0 +1,75 @@
+import os
+import logging
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+log = logging.getLogger(__name__)
+
+"""
+Based on environment variables, create the data store connection engine.
+:return:
+"""
+data_store_type = os.environ.get('process_tracking_data_store_type')
+data_store_username = os.environ.get('process_tracking_data_store_username')
+data_store_password = os.environ.get('process_tracking_data_store_password')
+data_store_host = os.environ.get('process_tracking_data_store_host')
+data_store_port = os.environ.get('process_tracking_data_store_port')
+data_store_name = os.environ.get('process_tracking_data_store_name')
+
+data_store_error_flag = False
+
+if data_store_type is None:
+    raise Exception('Data store type is not set.')
+    data_store_error_flag = True
+
+if data_store_username is None:
+    raise Exception('Data store username is not set.')
+    data_store_error_flag = True
+
+if data_store_password is None:
+    raise Exception('Data store password is not set')
+    data_store_error_flag = True
+
+if data_store_host is None:
+    raise Exception('Data store host is not set')
+    data_store_error_flag = True
+
+if data_store_port is None:
+    raise Exception('Data store port is not set')
+    data_store_error_flag = True
+
+if data_store_name is None:
+    raise Exception('Data store name is not set')
+    data_store_error_flag = True
+
+if data_store_error_flag:
+    raise Exception('Data store has not been properly configured.  Please read how to set up the Process '
+                    'Tracking data store by going to: <insert read the docs url here>')
+
+relational_stores = ['postgresql']
+nonrelational_stores = []
+
+supported_data_stores = relational_stores + nonrelational_stores
+
+if data_store_type in supported_data_stores:
+
+    if data_store_type in relational_stores:
+        if data_store_type == 'postgresql':
+            engine = create_engine(data_store_type + '://' + data_store_username + ':' + data_store_password
+                                   + '@' + data_store_host + '/' + data_store_name)
+
+            log.info("Attempting to connect to datastore %s, found at %s:%s" % (data_store_name
+                                                                                , data_store_host
+                                                                                , data_store_port))
+
+    elif data_store_type in nonrelational_stores:
+        Session = ''
+
+    Session = sessionmaker(bind=engine)
+
+    session = Session()
+    session.execute("SET search_path TO %s" % data_store_name)
+else:
+    raise Exception('Invalid data store type provided.  Please use: ' + ", ".join(supported_data_stores))
+    exit()