airflow envs and aspace index

- handles prod and stage airflow environments
- uses fixed aspace index in es as we won't reindex as often as we do with aleph (at least for now)

Author: JPrevost

docker-compose.yml

@@ -17,6 +17,7 @@ services:
       AIRFLOW__CORE__SQL_ALCHEMY_CONN: "postgresql+psycopg2://postgres@db/postgres"
       AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 30
       AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT: "False"
+      AIRFLOW_ENVIRONMENT: "stage"
       AWS_ACCESS_KEY_ID: "${AWS_ACCESS_KEY_ID}"
       AWS_SECRET_ACCESS_KEY: "${AWS_SECRET_ACCESS_KEY}"
       AWS_DEFAULT_REGION: "${AWS_DEFAULT_REGION}"
@@ -43,6 +44,7 @@ services:
       AIRFLOW__CORE__EXECUTOR: "CeleryExecutor"
       AIRFLOW__CELERY__BROKER_URL: "redis://redis:6379"
       AIRFLOW__CELERY__RESULT_BACKEND: "db+postgresql://postgres@db/postgres"
+      AIRFLOW_ENVIRONMENT: "stage"
       AWS_ACCESS_KEY_ID: "${AWS_ACCESS_KEY_ID}"
       AWS_SECRET_ACCESS_KEY: "${AWS_SECRET_ACCESS_KEY}"
       AWS_DEFAULT_REGION: "${AWS_DEFAULT_REGION}"
@@ -69,6 +71,7 @@ services:
       AIRFLOW__CORE__EXECUTOR: "CeleryExecutor"
       AIRFLOW__CELERY__BROKER_URL: "redis://redis:6379"
       AIRFLOW__CELERY__RESULT_BACKEND: "db+postgresql://postgres@db/postgres"
+      AIRFLOW_ENVIRONMENT: "stage"
       AWS_ACCESS_KEY_ID: "${AWS_ACCESS_KEY_ID}"
       AWS_SECRET_ACCESS_KEY: "${AWS_SECRET_ACCESS_KEY}"
       AWS_DEFAULT_REGION: "${AWS_DEFAULT_REGION}"

workflows/aspace_harvest.py

@@ -16,11 +16,12 @@
 network_config = json.loads(base64.b64decode(os.getenv('ECS_NETWORK_CONFIG')))
 cluster = os.getenv('ECS_CLUSTER')
 es_url = os.getenv('ES_URL')
+air_env = os.getenv('AIRFLOW_ENVIRONMENT')
 
 
 def set_s3():
     today = datetime.utcnow().strftime('%Y-%m-%d-%H-%M-%S')
-    s3_key = "s3://aspace-oai-s3-stage/{file}.xml".format(file=today)
+    s3_key = f"s3://aspace-oai-s3-{air_env}/{today}.xml"
     logging.info(s3_key)
     return s3_key
 
@@ -50,13 +51,12 @@ def check_if_records(**context):
 harvest = ECSOperator(task_id='harvest_step_1',
                       dag=dag,
                       cluster=cluster,
-                      task_definition='airflow-stage-oaiharvester',
+                      task_definition=f"airflow-{air_env}-oaiharvester",
                       overrides={'containerOverrides': [{
                                  'command': ["--out={{ task_instance.xcom_pull(task_ids='set_s3') }}",
                                              '--host=https://archivesspace.mit.edu/oai',
-                                             '--format=oai_ead',
-                                             '--verbose', ],
-                                 'name': 'airflow-stage-oaiharvester',
+                                             '--format=oai_ead', ],
+                                 'name': f"airflow-{air_env}-oaiharvester",
                                  }]},
                       network_configuration=network_config)
 
@@ -79,16 +79,17 @@ def check_if_records(**context):
 ingest = ECSOperator(task_id='harvest_step_3',
                      dag=dag,
                      cluster=cluster,
-                     task_definition='airflow-stage-mario',
+                     task_definition=f"airflow-{air_env}-mario",
                      overrides={'containerOverrides': [{
-                                  'command': ['--url=' + es_url,
+                                  'command': [f"--url={es_url}",
+                                              '--index=aspace_2019_12',
                                               'ingest',
                                               '--prefix=aspace',
                                               '--type=archives',
                                               '--auto',
                                               '--debug',
                                               "{{ task_instance.xcom_pull(task_ids='set_s3') }}", ],
-                                  'name': 'airflow-stage-mario',
+                                  'name': f"airflow-{air_env}-mario",
                                   }]},
                      network_configuration=network_config)