POST/ processes endpoint (#22)

* Adding IOClient and jobs results and payload endpoints

* Adding write object methods for IOClient

* Adding delete bucket methods for IOClient

* post processes endpoint

* use discriminant union for workflow configs

* addressing PR comments

* update status code

* fixup all the __root__ nonsense

* clean up process execute insert queries

* empty string not none for create/drop db name

* workflow name doesn't match process id check

* addressing PR comments

* remove id in payload/update tests

* dont use relative path in bad workflow config

---------

Co-authored-by: Hector Machin <hector.machin@ursaspace.com>
Co-authored-by: jkeifer <jkeifer@element84.com>

CONTRIBUTING.md

@@ -34,7 +34,7 @@ git commit -m "message" --no-verify
 Start the swoop api by running:
 
 ```commandline
-uvicorn swoop.api.main:app --host 0.0.0.0 --port 8000
+uvicorn swoop.api.main:app --host 0.0.0.0 --port 8000 --reload
 ```
 
 ## Testing

db/migrations/20230501205418_cache.sql

@@ -8,9 +8,9 @@ CREATE TABLE IF NOT EXISTS swoop.input_item (
 
 CREATE TABLE IF NOT EXISTS swoop.payload_cache (
   payload_uuid uuid DEFAULT gen_random_uuid() PRIMARY KEY,
-  payload_hash bytea,
-  workflow_version smallint,
-  workflow_name text,
+  payload_hash bytea UNIQUE,
+  workflow_version smallint NOT NULL,
+  workflow_name text NOT NULL,
   created_at timestamptz NOT NULL DEFAULT now(),
   invalid_after timestamptz
 );

db/schema.sql

@@ -39,9 +39,9 @@ INSERT INTO swoop.event_state (name, description) VALUES
 
 CREATE TABLE IF NOT EXISTS swoop.payload_cache (
   payload_uuid uuid DEFAULT gen_random_uuid() PRIMARY KEY,
-  payload_hash bytea,
-  workflow_version smallint,
-  workflow_name text,
+  payload_hash bytea UNIQUE,
+  workflow_version smallint NOT NULL,
+  workflow_name text NOT NULL,
   created_at timestamptz NOT NULL DEFAULT now(),
   invalid_after timestamptz
 );

pyproject.toml

@@ -62,3 +62,6 @@ readme = {file = "README.md"}
 
 [tool.isort]
 profile = "black"
+
+[tool.pytest.ini_options]
+minversion = "6.0"

src/swoop/api/config.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 from pydantic import BaseSettings, Field
 
 
@@ -37,7 +39,7 @@ def __init__(self, *args, _env_file=None, **kwargs):
     bucket_name: str
     execution_dir: str
     s3_endpoint: str
-    workflow_config_file: str
+    workflow_config_file: Path
 
     class Config:
         env_prefix = "swoop_"

src/swoop/api/exceptions.py

@@ -0,0 +1,6 @@
+class SwoopApiException(Exception):
+    pass
+
+
+class WorkflowConfigError(SwoopApiException, ValueError):
+    pass

src/swoop/api/models.py → src/swoop/api/models/__init__.py

@@ -30,11 +30,6 @@ class ConfClasses(BaseModel):
     conformsTo: list[str]
 
 
-class Response(Enum):
-    raw = "raw"
-    document = "document"
-
-
 class Type(Enum):
     process = "process"
 
@@ -191,8 +186,6 @@ class ProcessSummary(DescriptionType):
     outputTransmission: list[TransmissionMode] | None = None
     description: str | None = None
     handler: str | None = None
-    argoTemplate: str | None = None
-    cacheEnabled: bool | None = None
     cacheKeyHashIncludes: list[str] | None = None
     cacheKeyHashExcludes: list[str] | None = None
     links: list[Link] | None = None
@@ -211,16 +204,6 @@ class InlineOrRefData(BaseModel):
     __root__: InputValueNoObject | QualifiedInputValue | Link
 
 
-class Execute(BaseModel):
-    # TODO: I believe this is where we need to specify the input payload schema
-    inputs: dict[str, InlineOrRefData | list[InlineOrRefData]] | None = None
-    # TODO: We should likely omit the ability to specify outputs
-    outputs: dict[str, Output] | None = None
-    # TODO: Response isn't really to be supported, all results are json
-    response: Response | None = "raw"
-    subscriber: Subscriber | None = None
-
-
 class Results(BaseModel):
     # TODO: This becomes the schema of the workflow output
     __root__: dict[str, InlineOrRefData] | None = None

src/swoop/api/models/workflows.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from abc import ABC
+from enum import Enum
+from pathlib import Path
+from typing import Annotated, Literal, Union
+
+import yaml
+from pydantic import BaseModel, Field, StrictBool, StrictInt, StrictStr
+
+from swoop.api.exceptions import WorkflowConfigError
+
+
+class Response(Enum):
+    # raw = "raw"
+    document = "document"
+
+
+class BaseWorkflow(BaseModel, ABC):
+    name: StrictStr
+    description: StrictStr
+    version: StrictInt
+    cache_key_hash_includes: list[StrictStr] = []
+    cache_key_hash_excludes: list[StrictStr] = []
+
+
+class ArgoWorkflow(BaseWorkflow):
+    handler: Literal["argo-workflow"]
+    argo_template: StrictStr
+
+
+class CirrusWorkflow(BaseWorkflow):
+    handler: Literal["cirrus-workflow"]
+    sfn_arn: StrictStr
+
+
+Workflow = Annotated[
+    Union[
+        ArgoWorkflow,
+        CirrusWorkflow,
+    ],
+    Field(discriminator="handler"),
+]
+
+
+class Feature(BaseModel):
+    id: StrictStr
+    collection: StrictStr
+
+
+class UploadOptions(BaseModel):
+    path_template: StrictStr
+    collections: dict
+    public_assets: list[StrictStr] = []
+    headers: dict
+    s3_urls: StrictBool
+
+
+class Process(BaseModel):
+    description: StrictStr | None = None
+    tasks: dict
+    # input_collections: Optional[list[StrictStr]] = None
+    upload_options: UploadOptions
+    workflow: StrictStr
+
+
+class Payload(BaseModel):
+    type: StrictStr = "FeatureCollection"
+    features: list[Feature]
+    process: list[Process]
+
+
+class InputPayload(BaseModel):
+    payload: Payload
+
+
+class Execute(BaseModel):
+    # TODO: I believe this is where we need to specify the input payload schema
+    # inputs: dict[str, InlineOrRefData | list[InlineOrRefData]] | None = None
+    inputs: InputPayload
+    # TODO: We should likely omit the ability to specify outputs
+    # outputs: dict[str, Output] | None = None
+    # TODO: Response isn't really to be supported, all results are json
+    response: Literal["document"] = "document"
+    # subscriber: Subscriber | None = None
+
+
+class Workflows(dict[str, Workflow]):
+    class _type(BaseModel):
+        __root__: dict[str, Workflow]
+
+    @classmethod
+    def from_yaml(cls, path: Path) -> Workflows:
+        try:
+            workflows = yaml.safe_load(path.read_text())["workflows"]
+            for name, workflow in workflows.items():
+                workflow["name"] = name
+            return cls(cls._type.parse_obj(workflows).__root__)
+        except Exception as e:
+            raise WorkflowConfigError("Could not load workflow configuration") from e