loader (#56)

Makefile

@@ -5,11 +5,11 @@ gha-setup:
 
 .PHONY: init
 init: gha-setup
-	docker compose -f workflows/docker-compose.yml up -d
 	$(MAKE) -C entities local-init
+	$(MAKE) -C workflows local-init
 
 .PHONY: clean
 clean:
-	docker compose -f workflows/docker-compose.yml down
 	$(MAKE) -C entities local-clean
+	$(MAKE) -C workflows local-clean
 	docker compose down

entities/docker-compose.yml

@@ -2,7 +2,6 @@ version: "3.8"
 
 services:
   entities-db:
-    platform: linux/arm64
     image: postgres:15
     restart: always
     environment:
@@ -26,7 +25,6 @@ services:
       ["server", "--config", "/entities/config/config.yaml"]
   entities:
     image: "platformics-entities"
-    platform: linux/arm64
     build:
       context: "."
       args:

workflows/Makefile

@@ -31,7 +31,7 @@ rm-pycache: ## remove all __pycache__ files (run if encountering issues with pyc
 local-init:
 	$(docker_compose) up -d
 	while [ -z "$$($(docker_compose) exec -T postgres psql $(LOCAL_DB_CONN_STRING) -c 'select 1')" ]; do echo "waiting for db to start..."; sleep 1; done;
-	$(docker_compose) run $(FOLDER) make alembic-upgrade-head
+	$(docker_compose) run $(FOLDER) alembic upgrade head
 	$(MAKE) local-seed
 
 .PHONY: debugger
@@ -111,13 +111,13 @@ gha-setup:
 
 ### ALEMBIC #############################################
 alembic-upgrade-head:
-	alembic upgrade head ## Run alembic migrations locally
+	$(docker_compose) exec workflows alembic upgrade head ## Run alembic migrations locally
 
 alembic-undo-migration: ## Downgrade the latest alembic migration
-	alembic downgrade -1
+	$(docker_compose) exec workflows alembic downgrade -1
 
 alembic-autogenerate: ## Create new alembic migrations files based on SA schema changes.
-	alembic revision --autogenerate -m "$(MESSAGE)" --rev-id $$(date +%Y%m%d_%H%M%S)
+	$(docker_compose) exec workflows alembic revision --autogenerate -m "$(MESSAGE)" --rev-id $$(date +%Y%m%d_%H%M%S)
 
 ### SWIPE Plugin Tests
 

workflows/config.py

@@ -1,6 +1,6 @@
 from importlib.metadata import entry_points
 from typing import Dict
-from plugin_types import WorkflowRunner
+from plugin_types import EventBus, WorkflowRunner
 
 
 def load_workflow_runners() -> Dict[str, WorkflowRunner]:
@@ -10,3 +10,11 @@ def load_workflow_runners() -> Dict[str, WorkflowRunner]:
         assert isinstance(workflow_runner, WorkflowRunner)
         workflow_runners_by_name[plugin.name] = workflow_runner
     return workflow_runners_by_name
+
+def load_event_buses() -> Dict[str, EventBus]:
+    event_buses_by_name: Dict[str, EventBus] = {}
+    for plugin in entry_points(group="czid.plugin.event_bus"):
+        event_bus = plugin.load()()
+        assert isinstance(event_bus, EventBus)
+        event_buses_by_name[plugin.name] = event_bus
+    return event_buses_by_name

workflows/database/models/__init__.py

@@ -5,7 +5,5 @@
     RunStatus,
     Run,
     RunStep,
-    WorkflowVersionInput,
-    WorkflowVersionOutput,
     RunEntityInput,
 )  # noqa: F401

workflows/database/models/workflow.py

@@ -15,27 +15,17 @@ class Workflow(Base):
     name = Column(String, nullable=False)
     default_version = Column(String, nullable=False)
     minimum_supported_version = Column(String, nullable=False)
-    versions = relationship("WorkflowVersion", back_populates="workflow")
+    versions = relationship("WorkflowVersion", back_populates="workflow", foreign_keys="WorkflowVersion.workflow_id")
 
 
 class WorkflowVersion(Base):
     __tablename__ = "workflow_version"
     # TODO: replace with uuid7
     id = Column(Integer, primary_key=True, autoincrement=True)
-    version = Column(String, nullable=False)
-    type = Column(String, nullable=False)
-    package_uri = Column(String, nullable=False)
-    beta = Column(Boolean, default=False, nullable=False)
-    deprecated = Column(Boolean, default=False, nullable=False)
-    # TODO: add this back in when we add JSONB to strawberry-sqlalchemy-mapper
-    # graph_json = Column(JSONB)
-    graph_json = Column(String)
-    workflow_id = Column(Integer, ForeignKey("workflow.id"), nullable=False)
-    workflow = relationship("Workflow", back_populates="versions")
-    runs = relationship("Run", back_populates="workflow_version")
-    workflow_version_inputs = relationship("WorkflowVersionInput", back_populates="workflow_version")
-    workflow_version_outputs = relationship("WorkflowVersionOutput", back_populates="workflow_version")
-
+    workflow_id = Column(Integer, ForeignKey('workflow.id'), nullable=False)
+    workflow = relationship('Workflow', back_populates='versions', foreign_keys=[workflow_id])
+    runs = relationship('Run', back_populates='workflow_version', foreign_keys='Run.workflow_version_id')
+    manifest = Column(String, nullable=False)
 
 @strawberry.enum
 class RunStatus(enum.Enum):
@@ -60,12 +50,12 @@ class Run(Base):
     inputs_json = Column(String, nullable=False)
     # TODO: add this back in when we add JSONB to strawberry-sqlalchemy-mapper
     # outputs_json = Column(JSONB)
-    outputs_json = Column(String)
+    outputs_json = Column(String, nullable=True)
     status = Column(Enum(RunStatus), nullable=False, default=RunStatus.STARTED, name="status")
     workflow_version_id = Column(Integer, ForeignKey("workflow_version.id"), nullable=False)
-    workflow_version = relationship("WorkflowVersion", back_populates="runs")
-    run_steps = relationship("RunStep", back_populates="run")
-    run_entity_inputs = relationship("RunEntityInput", back_populates="run")
+    workflow_version = relationship("WorkflowVersion", back_populates="runs", foreign_keys=[workflow_version_id])
+    run_steps = relationship("RunStep", back_populates="run", foreign_keys="RunStep.run_id")
+    run_entity_inputs = relationship("RunEntityInput", back_populates="run", foreign_keys="RunEntityInput.run_id")
 
 
 @strawberry.enum
@@ -79,40 +69,18 @@ class RunStep(Base):
     # TODO: replace with uuid7
     id = Column(Integer, primary_key=True, autoincrement=True)
     run_id = Column(Integer, ForeignKey("run.id"), nullable=False)
-    run = relationship("Run", back_populates="run_steps")
+    run = relationship("Run", back_populates="run_steps", foreign_keys=[run_id])
     started_at = Column(DateTime, nullable=False, server_default=func.now())
     ended_at = Column(DateTime)
     status = Column(Enum(RunStepStatus), name="status")
 
-
-class WorkflowVersionInput(Base):
-    __tablename__ = "workflow_version_input"
-    # TODO: replace with uuid7
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    workflow_version_id = Column(Integer, ForeignKey("workflow_version.id"), nullable=False)
-    workflow_version = relationship("WorkflowVersion", back_populates="workflow_version_inputs")
-    name = Column(String, nullable=False)
-    entity_type = Column(String, nullable=False)
-    run_entity_inputs = relationship("RunEntityInput", back_populates="workflow_version_input")
-
-
-class WorkflowVersionOutput(Base):
-    __tablename__ = "workflow_version_output"
-    # TODO: replace with uuid7
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    workflow_version_id = Column(Integer, ForeignKey("workflow_version.id"), nullable=False)
-    workflow_version = relationship("WorkflowVersion", back_populates="workflow_version_outputs")
-    name = Column(String, nullable=False)
-    output_type = Column(String, nullable=False)
-    output_type_version = Column(String, nullable=False)
-
-
 class RunEntityInput(Base):
     __tablename__ = "run_entity_input"
     # TODO: replace with uuid7
     id = Column(Integer, primary_key=True, autoincrement=True)
-    run_id = Column(Integer, ForeignKey("run.id"), nullable=False)
-    run = relationship("Run", back_populates="run_entity_inputs")
-    workflow_version_input_id = Column(Integer, ForeignKey("workflow_version_input.id"), nullable=False)
-    workflow_version_input = relationship("WorkflowVersionInput", back_populates="run_entity_inputs")
+    run_id = Column(Integer, ForeignKey('run.id'), nullable=False)
+    run = relationship('Run', back_populates='run_entity_inputs', foreign_keys=[run_id])
+    # workflow_version_input_id = Column(Integer, ForeignKey('workflow_version_input.id'), nullable=False)
+    # workflow_version_input = relationship('WorkflowVersionInput', back_populates='run_entity_inputs')
     entity_id = Column(Integer, nullable=False)
+    field_name = Column(String, nullable=False)