Add Connector Registration and Dynamic Dispatch

.github/workflows/distribute_socrata_from_bytes.yml

@@ -1,5 +1,5 @@
-name: 📬 Distribute Socrata From Bytes
-run-name: "📬 Distribute Socrata From Bytes: ${{ inputs.PRODUCT_NAME }}-${{ inputs.DATASET_VERSION }}-${{ inputs.DATASET }}-${{ inputs.DESTINATION_ID }}"
+name: 📬 Distribute From Bytes
+run-name: "📬 Distribute From Bytes: ${{ inputs.PRODUCT_NAME }}-${{ inputs.DATASET_VERSION }}-${{ inputs.DATASET }}-${{ inputs.DESTINATION_ID }}"
 
 on:
   workflow_dispatch:
@@ -8,31 +8,40 @@ on:
         description: "Name of the product"
         type: string
         required: true
-      DATASET_VERSION:
+      PRODUCT_VERSION:
         description: "Version to push"
         type: string
         required: true
-      DESTINATION_ID:
-        description: "Destination ID (e.g. `colp_socrata`)"
-        type: string
-        required: true
       DATASET:
-        description: "Dataset to push (defaults to PRODUCT_NAME when omitted)"
+        description: "Dataset to push"
         type: string
-        required: false
+        required: true
       PUBLISH:
         description: "Publish the revision? (if not, will leave the revision open)"
         type: boolean
         default: false
-        required: true
-      IGNORE_VALIDATION_ERRORS:
-        description: "Ignore Validation Errors? (Will still perform validation, but will just log the outputs to the console)"
+        required: false
+      METADATA_ONLY:
+        description: "Distribute just the metadata?"
         type: boolean
         default: false
+        required: false
+      DESTINATION_ID_FILTER:
+        description: "Destination ID (e.g. `colp_socrata`)"
+        type: string
+        required: false
+      DESTINATION_TYPE_FILTER:
+        description: "Destination type (e.g. 'ftp')"
+        type: string
+        required: false
       SKIP_VALIDATION:
         description: "Skip validation altogether?"
         type: boolean
         default: false
+      IGNORE_VALIDATION_ERRORS:
+        description: "Ignore Validation Errors? (Will still perform validation, but will just log the outputs to the console)"
+        type: boolean
+        default: false
 jobs:
   publish:
     runs-on: ubuntu-22.04
@@ -43,9 +52,17 @@ jobs:
       image: nycplanning/build-base:latest
     env:
       PUBLISHING_BUCKET: edm-publishing
+      RECIPES_BUCKET: edm-recipes
+      PRODUCT_METADATA_REPO_PATH: product-metadata
+      _TYPER_STANDARD_TRACEBACK: 1
     steps:
       - uses: actions/checkout@v4
 
+      - uses: actions/checkout@v4
+        with:
+          repository: NYCPlanning/product-metadata
+          path: product-metadata
+
       - name: Load Secrets
         uses: 1password/load-secrets-action@v1
         with:
@@ -58,3 +75,20 @@ jobs:
           SOCRATA_USER: "op://Data Engineering/DCP_OpenData/username"
           SOCRATA_PASSWORD: "op://Data Engineering/DCP_OpenData/password"
 
+
+      - name: Finish container setup
+        run: ./bash/docker_container_setup.sh
+
+      - name: Distribute to Destinations
+        run: |
+          python3 -m dcpy.cli lifecycle scripts package_and_distribute \
+            ${{ inputs.PRODUCT_NAME }} \
+            ${{ inputs.PRODUCT_VERSION }} \
+            ${{ inputs.DATASET }} \
+            bytes \
+            $(if [ ${{ inputs.DESTINATION_ID_FILTER }} != "" ]; then echo "-a ${{ inputs.DESTINATION_ID_FILTER }}"; fi) \
+            $(if [ ${{ inputs.DESTINATION_TYPE_FILTER }} != "" ]; then echo "-e ${{ inputs.DESTINATION_TYPE_FILTER }}"; fi) \
+            $(if [ ${{ inputs.SKIP_VALIDATION }} = "true" ]; then echo "-y"; fi) \
+            $(if [ ${{ inputs.IGNORE_VALIDATION_ERRORS }} = "true" ]; then echo "-i"; fi) \
+            $(if [ ${{ inputs.PUBLISH }} = "true" ]; then echo "-p"; fi) \
+            $(if [ ${{ inputs.METADATA_ONLY }} = "true" ]; then echo "-m"; fi)

.gitignore

@@ -1,7 +1,7 @@
 # Lifecycle artifacts
 .library/
 .publishing/
-.package/
+.lifecycle/
 output/
 .output/
 .data

dcpy/connectors/ftp.py

@@ -0,0 +1,6 @@
+class FTPConnector:
+    def push(self, dest_path: str, ftp_profile: str):
+        raise Exception("Push not implemented for FTP")
+
+    def pull(self, **kwargs):
+        raise Exception("Pull not implemented for FTP")

dcpy/connectors/socrata/publish.py

@@ -578,10 +578,10 @@ def __init__(self, metadata: md.Metadata, destination_id: str):
 
 
 def push_dataset(
+    *,
     metadata: md.Metadata,
     dataset_destination_id: str,
     dataset_package_path: Path,
-    *,
     publish: bool = False,
     metadata_only: bool = False,
 ):

dcpy/lifecycle/__init__.py

@@ -1,5 +1,7 @@
 from pathlib import Path
 
+BASE_PATH = Path(".lifecycle")
+
 
 class WORKING_DIRECTORIES:
-    packaging = Path(".package")
+    packaging = BASE_PATH / Path("package")

dcpy/lifecycle/distribute/__init__.py

@@ -0,0 +1,44 @@
+from typing import Unpack
+
+from dcpy.lifecycle.distribute.connectors import (
+    DistributionSFTPConnector,
+    SocrataPublishConnector,
+)
+from dcpy.models.lifecycle.distribute import (
+    DatasetDestinationPushArgs,
+    DistributeResult,
+)
+from dcpy.models.connectors import ConnectorDispatcher
+
+
+# Register all default connectors for `lifecycle.distribute`.
+# Third parties can similarly register their own connectors,
+# so long as the connector implements a ConnectorDispatcher protocol.
+dispatcher = ConnectorDispatcher[DatasetDestinationPushArgs, dict]()
+
+dispatcher.register(conn_type="socrata", connector=SocrataPublishConnector())
+dispatcher.register(conn_type="sftp", connector=DistributionSFTPConnector())
+
+
+def to_dataset_destination(
+    **push_kwargs: Unpack[DatasetDestinationPushArgs],
+) -> DistributeResult:
+    """Distribute a dataset and specific dataset_destination_id.
+
+    Requires fully rendered template, ie there should be no template variables in the metadata
+    """
+    ds_md = push_kwargs["metadata"]
+    dest = ds_md.get_destination(push_kwargs["dataset_destination_id"])
+    dest_type = dest.type
+
+    try:
+        result = dispatcher.push(dest_type, push_kwargs)
+        return DistributeResult.from_push_kwargs(
+            result=result, success=True, push_args=push_kwargs
+        )
+    except Exception as e:
+        return DistributeResult.from_push_kwargs(
+            result=f"Error pushing {push_kwargs['metadata'].attributes.display_name} to dest_type: {dest_type}, destination: {dest.id}: {str(e)}",
+            success=False,
+            push_args=push_kwargs,
+        )

dcpy/lifecycle/distribute/_cli.py

@@ -1,7 +1,41 @@
 import typer
+from pathlib import Path
 
-from dcpy.lifecycle.distribute import socrata
+from dcpy.lifecycle import distribute
+import dcpy.models.product.dataset.metadata as m
 
 app = typer.Typer()
 
-app.add_typer(socrata.socrata_app, name="socrata")
+
+@app.command("from_local")
+def _dist_from_local(
+    package_path: Path = typer.Argument(),
+    dataset_destination_id: str = typer.Argument(),
+    metadata_path: Path = typer.Option(
+        None,
+        "-m",
+        "--metadata-path",
+        help="(Optional) Metadata Path Override",
+    ),
+    publish: bool = typer.Option(
+        False,
+        "-p",
+        "--publish",
+        help="Publish the Revision? Or leave it open.",
+    ),
+    metadata_only: bool = typer.Option(
+        False,
+        "-z",
+        "--metadata-only",
+        help="Only push metadata (including attachments).",
+    ),
+):
+    md = m.Metadata.from_path(metadata_path or (package_path / "metadata.yml"))
+    result = distribute.to_dataset_destination(
+        metadata=md,
+        dataset_destination_id=dataset_destination_id,
+        publish=publish,
+        dataset_package_path=package_path,
+        metadata_only=metadata_only,
+    )
+    print(result)

dcpy/lifecycle/distribute/connectors.py

@@ -0,0 +1,42 @@
+from typing import Any
+
+from dcpy.connectors.ftp import FTPConnector
+from dcpy.connectors.socrata import publish as socrata_pub
+from dcpy.models.lifecycle.distribute import DatasetDestinationPushArgs
+
+# Sadly, can't use Unpack on kwarg generics yet.
+# https://github.com/python/typing/issues/1399
+
+
+# Wrap the connectors to bind them to the `PublisherPushKwargs`
+# so that we can register and delegate calls.
+# This is the recommended way for third parties to add custom Distribution Connectors.
+class DistributionSFTPConnector:
+    conn_type: str
+
+    def __init__(self):
+        self.conn_type = "sftp"
+        self._base_connector = FTPConnector()
+
+    def push(self, arg: DatasetDestinationPushArgs) -> Any:
+        md = arg["metadata"]
+        dest = md.get_destination(arg["dataset_destination_id"])
+        dest_path = dest.custom["destination_path"]
+        user_id = dest.custom["user_id"]
+        self._base_connector.push(dest_path=dest_path, ftp_profile=user_id)
+
+    def pull(self, _: dict) -> Any:
+        raise Exception("Pull is not defined for any Distribution Connectors.")
+
+
+class SocrataPublishConnector:
+    conn_type = "socrata"
+
+    def push(
+        self,
+        arg: DatasetDestinationPushArgs,
+    ) -> Any:
+        return socrata_pub.push_dataset(**arg)
+
+    def pull(self, _: dict):
+        raise Exception("Pull not implemented for Socrata Connector")