Building models for Process Tracking.

Author: OpenDataAlex

.idea/vcs.xml

@@ -7,6 +7,9 @@ verify_ssl = true
 
 [packages]
 sqlalchemy = "*"
+sqlalchemy-utils = "*"
+dateutil = "*"
+python-dateutil = "*"
 
 [requires]
 python_version = "3.7"

Pipfile

@@ -0,0 +1,34 @@
+# 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.orm import relationship
+from sqlalchemy_utils.types import uuid
+
+Base = declarative_base()
+default_date = parser.parse('1900-01-01 00:00:00')
+
+
+class ProcessType(Base):
+    __tablename__ = 'process_type'
+
+
+
+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)
+    total_record_count = Column(Integer, nullable=False, default=0)
+    latest_run_actor_id = Column()
+    process_type_id = Column()
+    last_failed_run_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)

Pipfile.lock

@@ -4,6 +4,7 @@
 from source import Source
 from tool import Tool
 
+
 class ProcessTracker:
 
     def __init__(self, process_name, actor_name, tool_name, source_name):
@@ -15,6 +16,6 @@ def __init__(self, process_name, actor_name, tool_name, source_name):
 
     def register_new_process_run(self):
         """
-        
+
         :return:
         """