set-schema: fix

backoffice/backoffice/workflows/airflow_utils.py

@@ -29,7 +29,7 @@ def trigger_airflow_dag(dag_id, workflow_id, extra_data=None):
     data = {"dag_run_id": str(workflow_id), "conf": {"workflow_id": str(workflow_id)}}
 
     if extra_data is not None:
-        data["conf"].update(extra_data)
+        data["conf"]["data"] = extra_data
 
     url = f"{AIRFLOW_BASE_URL}/api/v1/dags/{dag_id}/dagRuns"
 

workflows/dags/author/author_create/author_create_init.py

@@ -2,6 +2,7 @@
 import logging
 
 from airflow.decorators import dag, task
+from airflow.models import Variable
 from airflow.models.param import Param
 from hooks.backoffice.workflow_management_hook import AUTHORS, WorkflowManagementHook
 from hooks.backoffice.workflow_ticket_management_hook import (
@@ -50,10 +51,12 @@ def set_workflow_status_to_running(**context):
 
     @task()
     def set_schema(**context):
-        schema = "https://inspirehep.net/schemas/records/authors.json"
+        schema = Variable.get("author_schema")
         workflow_management_hook.partial_update_workflow(
             workflow_id=context["params"]["workflow_id"],
-            workflow_partial_update_data={"data": {"$schema": schema}},
+            workflow_partial_update_data={
+                "data": {**context["params"]["data"], "$schema": schema}
+            },
             collection=AUTHORS,
         )
 

workflows/scripts/variables/variables.json

@@ -1,4 +1,5 @@
 {
   "backoffice_token": "2e04111a61e8f5ba6ecec52af21bbb9e81732085",
-  "inspire_token": "CHANGE_ME"
+  "inspire_token": "CHANGE_ME",
+  "author_schema": "https://inspirehep.net/schemas/records/authors.json"
 }