Skip to content

DAGS

Jump to bottom
amy wieliczka edited this page Jun 23, 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 parameters

  • Calling trigger_dag in the Django 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

    Invokes the DAG via the rest api, returns the URL to view the dag run in the airflow UI. If track_dag=True, has a side effect of creating a 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 Django Shell for a specific instance of a FindingAid object from cincoctrl.findingaids.models with parameters force_publish=False.
    Updates the EAD with supplementary files, queues the status of the finding aid: setting the finding aid's status to either "queued_publish" (if current status is "publish") or "queued_preview" (otherwise), and then calls 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}"

  • In views.py,

    • Submitting the FindingAidCreateView or the FindingAidUpdateView (both inherit EADMixin, which defines form_valid(self, form: ModelForm): if the submitted form instance has an ead_file, update the ead with supplementary files, call queue_index() on the form instance, return response.)
    • Submitting the RecordExpressCreateView or the RecordExpressUpdateView (both inherit RecordExpressMixin which defines 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))
    • Visiting the PublishRecordView, which defines 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
    • Visiting the PreviewRecordView, which defines get_object(self, **kwargs): if objects has an ead_file, update the ead with supplementary files, call queue_index() on the object, return object
    • Submitting the AttachPDFView, which defines 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.

  • In signals.py

    • On post_save of a SupplementaryFile, trigger_reindex(): if instance.textract_status == "SUCCEEDED" then queue_index() for the instance's related finding aid

  • In 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

When the trigger_dag() function is called, a JobTrigger is created. CincoCtrl has a script to poll for the status of JobTriggers and JobRuns. It looks for all JobTriggers with a non-null dag_run_id, excluding any JobTriggers with associated JobRuns (1 to many JobTrigger to JobRuns) with status of SUCCEEDED or FAILED. It also looks for all JobRuns, excluding any JobRuns with status SUCCEEDED or FAILED.

For each of the Jobs (Triggers or Runs), it calls update_job_run, which gets or creates a JobRun for the associated Job and deletes the JobTrigger.

If a JobRun ends up in a FAILED state and someone re-runs the DAG from the airflow UI, the JobRun is stuck in a FAILED state, regardless of the state of the re-ran DAG.

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