Skip to content

DAGS

Jump to bottom
amy wieliczka edited this page Jun 22, 2026 · 6 revisions

There are, at the time of writing, 6 DAGS. 3 DAGS that operate on a single finding aid, 2 DAGS that poll for status updates, and 2 DAGS that operate on a bulk set of finding aids.

Index Finding Aid DAG

Params:

  • finding_aid_id: type="integer", description="The ID of the Finding Aid in CincoCtrl"
  • repository_code: default="", type="string", description="The repository code for the Finding Aid"
  • finding_aid_ark: default="", type="string", description="The ARK of the Finding Aid"
  • eadid: default="", type="string", description="If present the filename in s3 else the collection number"
  • preview: default="publish", type="string", description="Either preview or publish"
  • "cinco_environment": default="stage", enum=["stage", "prd"], description="The CincoCtrl and ArcLight environment to run"

Tasks:

  1. Make s3 key for temporary working directory s3_key = f"indexing/{finding_aid_id}/{datetime.now().isoformat()}"
  2. Prepare finding aid in CincoCtrl container: python manage.py prepare_finding_aid --finding_aid_id {finding_aid_id} --s3_key {s3_key}
    1. get or create the ead file (via express_record.html template) and copy into the s3 working directory
    2. create an indexing_env.sh file in the s3 working directory with the finding aid id, repository id, finding aid ark, and action (publish or preview)
    3. combine the textract output from all supplementary files as available
    4. save the extracted supplementary file text to the s3 working directory
  3. Index finding aid in ArcLight container bin/index-from-s3 {finding_aid_id} {s3_key} {repository_code} {finding_aid_ark} {eadid} {preview}
    1. sync the s3 working directory to /tmp/<finding_aid_id>
    2. set repository id and preview environment variables
    3. run bundle exect traject -I lib/ -u $SOLR_WRITER -i xml -c lib/arclight/traject/ead2_config.rb /tmp/<finding_aid_id>/finding-aid.xml -s ark=<finding_aid_ark> -s eadid=<eadid> -s preview=$PREVIEW
    4. cleanup the /tmp/ directory
  4. Clean Up S3 working directory delete_results = bucket.objects.filter(Prefix=prefix).delete()
  5. Request static finding aid rebuild in ArcLight container bin/generate-static-findaid {finding_aid_ark}
    1. run bundle exec rake static_finding_aid:generate[<finding_aid_ark>]
      1. StaticFindingAidRenderJob.new.perform(id)
  6. Wait 1 minute with a TimeDeltaSensor
  7. Clear Cloudfront cache:
invalidation_paths = [
    f"/findaid/{finding_aid_ark}*", 
    f"/findaid/static/{finding_aid_ark}*"
]

Triggering the Index Finding Aid DAG

The Index Finding Aid DAG can be triggered by:

  • Running it directly in the Airflow UI with the parameters specified
  • Calling trigger_dag in the Python Shell from cincoctrl.airflow_client.mwaa_api_client with parameters dag, dag_conf, client, related_models=None, dag_run_prefix=None, dag_note=None, *, track_dag=True: Invoke the DAG via the rest api, return the URL to view the dag run in the airflow UI. If track_dag=True, has a side effect of creating JobTrigger object with dag_id, dag_run_conf, airflow_url, dag_run_id, logical_date, rest_api_status_code, rest_api_resp
  • Calling FindingAid.queue_index in the Python Shell for a specific instance of a FindingAid object from cincoctrl.findingaids.models with parameters force_publish=False: call trigger_dag with parameters dag=index_finding_aid, dag_run_conf={"finding_aid_id": self.id, "repository_code"=self.repository.code, "finding_aid_ark"=self.ark, "eadid"=self.eadid, "preview"=action, "cinco_environment"=settings.CINCO_ENVIRONMENT), dag_note="Indexing {self.ark} from {self.repository.name} ({self.repository.code})", related_models=[self], dag_run_prefix="{[cinco-stage|cinco-prd]}__{ark}"), has a side effect of updating the ead with supplementary files and queuing the status of the finding aid (setting the finding aid's status field to either "queued_publish" (if current status is "publish") or "queued_preview" (otherwise).
  • views.py
    • EADMixin.form_valid(self, form: ModelForm), if form instance has an ead_file, update the ead with supplementary files, call queue_index() on the form instance, return response.
    • RecordExpressMixin.form_valid(self, form: ModelForm), if form and inline formsets are valid, set finding aid defaults for the form, save it and all inline formsets, call queue_index() on the express record, and return super().form_valid(form)
    • PublishRecordView.get_object(self, **kwargs), if object has an ead_file, update the ead with supplementary files, call queue_index() on the object with force_publish=True, return object
    • PreviewRecordView.get_object(self, **kwargs), if objects has an ead_file, update the ead with supplementary files, call queue_index() on the object, return object
    • AttachPDFView.form_valid(self, form: ModelForm), if formset is valid, save formset, if form instance has an ead_file, update the ead with supplementary files, if an update has occurred, call queue_index.
  • signals.py
    • trigger_reindex(): on post_save of a SupplementaryFile, if instance.textract_status == "SUCCEEDED" then queue_index() for the instance's related finding aid
  • admin.py
    • FindingAidAdmin.index_finding_aid_action(): for finding aid in set, queue_index, return message with airflow urls for each finding aid in set.

Tracking Status of DAG Run

Tracking Solr Index Status of Finding Aid

Delete Finding Aid DAG

  1. Remove from index in ArcLight container: bin/remove-from-solr {finding_aid_ark} {repository_code}
  2. Remove from database in CincoCtrl container: python manage.py remove_finding_aid --ark {finding_aid_ark}
  3. Remove static finding aids from S3:
paths = [
    f"static_findaids/static_findaids/{ark}",
    f"static_findaids/oac4/{ark}",
    f"static_findaids/oac5/{ark}",
]

Unpublish Finding Aid DAG

  1. Remove from index in ArcLight container: bin/remove-from-solr {finding_aid_ark} {repository_code}
  2. Mark unpublished in CincoCtrl container: python manage.py mark_unpublished --ark {finding_aid_ark}

2 Polling DAGS

Poll Airflow Api

  1. Poll Airflow API from Stage CincoCtrl Container: python manage.py poll_airflow
  2. Poll Airflow API from Prod CincoCtrl Container: python manage.py poll_airflow

Poll Message Queue

  1. Poll SQS Message Queue from Stage CincoCtrl Container: python manage.py poll_sqs
  2. Poll SQS Message Queue from Prod CincoCtrl Container: python manage.py poll_sqs

2 Bulk DAGS

Bulk Index Finding Aids DAG

TODO: Have not added static finding aid rebuild to this DAG

  1. Make s3 key for temporary working directory s3_key = f"indexing/bulk/{s3_prefix}"
  2. Bulk prepare finding aids in CincoCtrl Container: python manage.py bulk_prepare_finding_aids --filters {queryset_filters} --s3_key {s3_key} --max_num_records {max_num_records} --max_file_size_in_MB {max_file_size_in_MB}
  3. Get list of finding aid batches from s3 [batch.get("Prefix").replace("media/", "").strip("/") for batch in s3.list_objects_v2(Bucket=bucket_name, Prefix=f"media/{s3_key}/", Delimiter="/").get('CommonPrefixes', [])]
  4. Run bulk indexing and s3 Cleanup as a task group in parallel for each batch of finding aids a. Bulk index batch of finding aids in Arclight container: bin/bulk-index-from-s3 {batch_s3_key} b. Cleanup s3 delete_results = bucket.objects.filter(Prefix=prefix).delete()

Bulk Delete Finding Aids DAG

  1. Bulk remove finding aids from solr in ArcLight container: bin/bulk-remove-from-solr {s3_key}
  2. Bulk remove finding aids from database in CincoCtrl container: python manage.py bulk_remove_finding_aids --s3_key {s3_key}
  3. Bulk remove static finding aids from s3:
removals = s3.get_object(Bucket=bucket_name, Key=s3_key)["Body"].read().decode("utf-8").splitlines()
for line in removals:
    ark, _ = line.split(",", 1)
    prefixes = [
        f"static_findaids/static_findaids/{ark}",
        f"static_findaids/oac4/{ark}",
        f"static_findaids/oac5/{ark}",
    ]
    for prefix in prefixes:
        delete_results = bucket.objects.filter(Prefix=prefix).delete()

Clone this wiki locally