Add datacontract publish command

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - datacontract catalog: Search form
 - `datacontract import --format bigquery`: Import from BigQuery format
+- `datacontract publish`: Publish the data contract to the Data Mesh Manager
 
 ## [0.10.3] - 2024-05-05
 

datacontract/cli.py

@@ -11,11 +11,10 @@
 from typer.core import TyperGroup
 from typing_extensions import Annotated
 
-from datacontract.catalog.catalog import create_index_html, \
-    create_data_contract_html
+from datacontract.catalog.catalog import create_index_html, create_data_contract_html
 from datacontract.data_contract import DataContract
-from datacontract.init.download_datacontract_file import \
-    download_datacontract_file, FileExistsException
+from datacontract.init.download_datacontract_file import download_datacontract_file, FileExistsException
+from datacontract.publish.publish import publish_to_datamesh_manager
 
 console = Console()
 
@@ -163,7 +162,12 @@ class ExportFormat(str, Enum):
 @app.command()
 def export(
     format: Annotated[ExportFormat, typer.Option(help="The export format.")],
-    output: Annotated[Path, typer.Option(help="Specify the file path where the exported data will be saved. If no path is provided, the output will be printed to stdout.")] = None,
+    output: Annotated[
+        Path,
+        typer.Option(
+            help="Specify the file path where the exported data will be saved. If no path is provided, the output will be printed to stdout."
+        ),
+    ] = None,
     server: Annotated[str, typer.Option(help="The server name to export.")] = None,
     model: Annotated[
         str,
@@ -204,7 +208,7 @@ def export(
     if output is None:
         console.print(result, markup=False)
     else:
-        with output.open('w') as f:
+        with output.open("w") as f:
             f.write(result)
         console.print(f"Written result to {output}")
 
@@ -228,6 +232,20 @@ def import_(
     console.print(result.to_yaml())
 
 
+@app.command(name="publish")
+def publish(
+    location: Annotated[
+        str, typer.Argument(help="The location (url or path) of the data contract yaml.")
+    ] = "datacontract.yaml",
+):
+    """
+    Publish the data contract to the Data Mesh Manager
+    """
+    publish_to_datamesh_manager(
+        data_contract=DataContract(data_contract_file=location),
+    )
+
+
 @app.command(name="catalog")
 def catalog(
     files: Annotated[

datacontract/publish/publish.py

@@ -0,0 +1,32 @@
+import os
+
+import requests
+
+from datacontract.data_contract import DataContract
+
+
+def publish_to_datamesh_manager(data_contract: DataContract):
+    try:
+        headers = {"Content-Type": "application/json", "x-api-key": _require_datamesh_manager_api_key()}
+        spec = data_contract.get_data_contract_specification()
+        id = spec.id
+        url = "https://api.datamesh-manager.com/api/datacontracts/{0}".format(id)
+        request_body = spec.model_dump_json().encode("utf-8")
+        response = requests.put(
+            url=url,
+            data=request_body,
+            headers=headers,
+        )
+        if response.status_code != 200:
+            print(f"Error publishing data contract to Data Mesh Manager: {response.text}")
+            exit(1)
+        print(f"Published data contract to {url}")
+    except Exception as e:
+        print(f"Failed publishing data contract. Error: {str(e)}")
+
+
+def _require_datamesh_manager_api_key():
+    datamesh_manager_api_key = os.getenv("DATAMESH_MANAGER_API_KEY")
+    if datamesh_manager_api_key is None:
+        raise Exception("Cannot publish data contract, as DATAMESH_MANAGER_API_KEY is not set")
+    return datamesh_manager_api_key