如何在Airflow REST API URL中正确处理自动生成的dag_run_id

问题原因 #

Airflow 2.2.5的REST API在解析路径参数时，无法正确将%2B解码回+。你使用urllib.parse.quote_plus()会把dag_run_id中的时区符号+编码成%2B，导致API搜索的dag_run_id是带有%2B的错误字符串，从而返回404。

解决方案 #

有两种可行的修正方式：

1. 直接使用原始dag_run_id字符串
   URL路径中的+字符本身是合法的，无需编码。直接把从/dags/{dag_id}/dagRuns接口获取的原始dag_run_id拼接到URL中即可：

   import requests
   
   api_base_url = 'http://myairflow:8080/api/v1'
   auth = ('username', 'password')
   
   dag_id = 'mydag'
   
   dag_runs_resp = requests.get(
       f'{api_base_url}/dags/{dag_id}/dagRuns', 
       auth=auth
   )
   last_dag_run_id = dag_runs_resp.json()['dag_runs'][-1]['dag_run_id']
   
   # 直接使用原始dag_run_id，无需编码
   dag_run_resp = requests.get(
       f'{api_base_url}/dags/{dag_id}/dagRuns/{last_dag_run_id}', 
       auth=auth
   ).json()
   
   print(dag_run_resp)
   

2. 使用urllib.parse.quote()替代quote_plus()
   如果需要处理dag_run_id中其他可能的特殊字符，使用quote()而非quote_plus()——它不会将+编码为%2B：

   import requests
   import urllib
   
   api_base_url = 'http://myairflow:8080/api/v1'
   auth = ('username', 'password')
   
   dag_id = 'mydag'
   
   dag_runs_resp = requests.get(
       f'{api_base_url}/dags/{dag_id}/dagRuns', 
       auth=auth
   )
   last_dag_run_id = dag_runs_resp.json()['dag_runs'][-1]['dag_run_id']
   
   # 使用quote()编码，保留原始的+符号
   encoded_dag_run_id = urllib.parse.quote(last_dag_run_id)
   
   dag_run_resp = requests.get(
       f'{api_base_url}/dags/{dag_id}/dagRuns/{encoded_dag_run_id}', 
       auth=auth
   ).json()
   
   print(dag_run_resp)
   

补充说明 #

quote_plus()主要用于处理URL查询参数（?key=value部分），它会把空格替换为+，同时编码特殊字符；而路径参数中不需要这种转换，+可以直接保留，因此使用quote()或者直接用原始字符串更适配Airflow 2.2.5的API行为。

内容的提问来源于stack exchange，提问作者chaooder