Trigger digital bookplate instances DAG when there is new bookplates data

[shelleydoljack]

When there are new bookplates identified from the fetch_digital_bookplates DAG, we want to automatically trigger the DAG that finds instances paid on bookplate funds. The data passed to trigger the dag should be a list of new funds (bookplate metadata structure) and a date of beginning of time on FOLIO (date should be 8/28/2023; there are 3 paid invoices prior to 9/1/2023).

When the DAG has this configuration:

@dag(
    default_args=default_args,
    schedule=CronDataIntervalTimetable(
        cron="0 2 * * WED", timezone="America/Los_Angeles"
    ),
    start_date=datetime(2023, 8, 28),
    catchup=False,
    tags=["digital bookplates"],
)
def digital_bookplate_instances():

And it is triggered w/out a config, it creates dag_runs that have data_interval_end and data_interval_start dates we can use on a weekly basis:

{
  "dag_runs": [
    {
      "conf": {},
      "dag_id": "digital_bookplate_instances",
      "dag_run_id": "manual__2024-09-19T20:23:18.515617+00:00",
      "data_interval_end": "2024-09-18T09:00:00+00:00",
      "data_interval_start": "2024-09-11T09:00:00+00:00",
      "end_date": null,
      "execution_date": "2024-09-19T20:23:18.515617+00:00",
      "external_trigger": true,
      "last_scheduling_decision": "2024-09-19T20:23:31.696585+00:00",
      "logical_date": "2024-09-19T20:23:18.515617+00:00",
      "note": null,
      "run_type": "manual",
      "start_date": "2024-09-19T20:23:18.601494+00:00",
      "state": "running"
    },
    {
      "conf": {},
      "dag_id": "digital_bookplate_instances",
      "dag_run_id": "scheduled__2024-09-11T09:00:00+00:00",
      "data_interval_end": "2024-09-18T09:00:00+00:00",
      "data_interval_start": "2024-09-11T09:00:00+00:00",
      "end_date": null,
      "execution_date": "2024-09-11T09:00:00+00:00",
      "external_trigger": false,
      "last_scheduling_decision": "2024-09-19T20:23:31.693525+00:00",
      "logical_date": "2024-09-11T09:00:00+00:00",
      "note": null,
      "run_type": "scheduled",
      "start_date": "2024-09-19T20:23:18.601346+00:00",
      "state": "running"
    }
  ],
  "total_entries": 2
}

But for dag runs for new bookplates, we need to trigger the dag with a configuration. When triggering the above defined dag with a logical_date value of 8/28/2023 00:00:00, the dag run that gets created is this:

{
      "conf": {},
      "dag_id": "digital_bookplate_instances",
      "dag_run_id": "manual__2023-08-28T00:00:00+00:00",
      "data_interval_end": "2023-08-23T09:00:00+00:00",
      "data_interval_start": "2023-08-16T09:00:00+00:00",
      "end_date": null,
      "execution_date": "2023-08-28T00:00:00+00:00",
      "external_trigger": true,
      "last_scheduling_decision": "2024-09-19T20:26:19.583010+00:00",
      "logical_date": "2023-08-28T00:00:00+00:00",
      "note": null,
      "run_type": "manual",
      "start_date": "2024-09-19T20:26:10.373003+00:00",
      "state": "running"
    }

We should use the logical_date field (execution_date is deprecated (See Release Notes Miscellaneous)) when the run_id starts with "manual_" for the invoice query params paymentDate>=logical_date

[shelleydoljack]

I just re-reviewed this ticket and it still contains accurate requirements.