timkpaine opened a new pull request, #64436: URL: https://github.com/apache/airflow/pull/64436
See discussion: https://github.com/apache/airflow/discussions/61800 This PR adds a native local REST client for "secure" contexts, e.g. plugins and inside DAG processing, to be able to access metadata that was previously available via direct DB access in Airflow 2. An example of this is e.g. a DAG that ensures a Pool exists before using it. This allows one to implement things dropped for Airflow 3 like e.g. the [Astronmer DAG cleanup workflow](https://www.astronomer.io/docs/learn/2.x/cleanup-dag-tutorial/). For example here is my variant now for Airflow 3 using the rest client: ```python def clean_dag_runs(delete_successful, delete_failed, mark_failed_as_successful, max_dagruns, days_to_keep, **kwargs): from airflow.api.client import get_local_rest_client client = get_local_rest_client(process_type="cleanup") utc_now = datetime.now(tz=UTC) cutoff_date = utc_now - timedelta(days=days_to_keep) _log.info(f"Cutoff date for clean: {cutoff_date}") all_dags = _list_all_dags(client) dag_ids = [d.dag_id for d in all_dags] _log.info(f"Found DAGs to clean up: {dag_ids}") deleted = 0 for dag_id in dag_ids: _log.info(f"Cleaning up DAG: {dag_id}") runs = _list_all_dag_runs(client, dag_id) # Filter by state if delete_successful is False: _log.info(f"Not deleting successful DAG runs for DAG: {dag_id}") runs = [r for r in runs if str(r.state) != "success"] if delete_failed is False: _log.info(f"Not deleting failed DAG runs for DAG: {dag_id}") runs = [r for r in runs if str(r.state) != "failed"] # Sort by logical_date ascending (use datetime.min as fallback for None) _min_dt = datetime.min.replace(tzinfo=UTC) runs.sort(key=lambda r: r.logical_date or _min_dt) total_runs = len(runs) _log.info(f"Found {total_runs} DAG runs to clean up for DAG: {dag_id}") for dr in runs: logical_date = dr.logical_date if (logical_date and logical_date < cutoff_date) or total_runs > max_dagruns: _log.info(f"Deleting DAG run: {dr.dag_run_id}") client.dag_runs.delete(dag_id, dr.dag_run_id) deleted += 1 total_runs -= 1 elif mark_failed_as_successful: if str(dr.state) == "failed": _log.info(f"Marking failed DAG run as successful: {dr.dag_run_id}") client.dag_runs.update(dag_id, dr.dag_run_id, state="success") elif not mark_failed_as_successful: break _log.info(f"Total DAG runs deleted: {deleted}") ``` There are a bunch of followup questions, principal amongst them is the questions of the API client itself and its distinction (?) from the public airflow rest client. ##### Was generative AI tooling used to co-author this PR? - [x] Yes (please specify the tool below) Copilot in VS Code -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
