我正在尝试使用SimpleHttpOperator来消费RESTful API。但是,正如其名称所示,它仅支持HTTP协议,而我需要消费HTTPS URI。因此,现在我必须使用Python的“requests”对象或从应用程序代码中处理调用。但这可能不是标准的方法。因此,我正在寻找其他可用选项,以从Airflow内部消费HTTPS URI。谢谢。
from airflow.operators.http_operator import SimpleHttpOperator
from airflow.exceptions import AirflowException
from operators.https_support.https_hook import HttpsHook
class HttpsOperator(SimpleHttpOperator):
def execute(self, context):
http = HttpsHook(self.method, http_conn_id=self.http_conn_id)
self.log.info("Calling HTTP method")
response = http.run(self.endpoint,
self.data,
self.headers,
self.extra_options)
if self.response_check:
if not self.response_check(response):
raise AirflowException("Response check returned False.")
if self.xcom_push_flag:
return response.text
钩子覆盖
from airflow.hooks.http_hook import HttpHook
import requests
class HttpsHook(HttpHook):
def get_conn(self, headers):
"""
Returns http session for use with requests. Supports https.
"""
conn = self.get_connection(self.http_conn_id)
session = requests.Session()
if "://" in conn.host:
self.base_url = conn.host
elif conn.schema:
self.base_url = conn.schema + "://" + conn.host
elif conn.conn_type: # https support
self.base_url = conn.conn_type + "://" + conn.host
else:
# schema defaults to HTTP
self.base_url = "http://" + conn.host
if conn.port:
self.base_url = self.base_url + ":" + str(conn.port) + "/"
if conn.login:
session.auth = (conn.login, conn.password)
if headers:
session.headers.update(headers)
return session
使用方法:
作为 SimpleHttpOperator 的替代品,可以直接使用。
Conn Id : sendgrid_templates_test
Conn Type : HTTP
Host : https://api.sendgrid.com/
Extra : { "authorization": "Bearer [my token]"}
然后在dag代码中:
get_templates = SimpleHttpOperator(
task_id='get_templates',
method='GET',
endpoint='/v3/templates',
http_conn_id = 'sendgrid_templates_test',
trigger_rule="all_done",
xcom_push=True
dag=dag,
)
我试过了,它起作用了。同时请注意,我的请求发生在分支操作之后,所以我需要适当设置触发规则(设置为“all_done”,以确保即使跳过其中一个分支也会触发),这与问题无关,但我想指出来。
现在要明确的是,由于我没有启用证书验证,因此我收到了不安全请求警告。但您可以在下面看到结果日志。
[2019-02-21 16:15:01,333] {http_operator.py:89} INFO - Calling HTTP method
[2019-02-21 16:15:01,336] {logging_mixin.py:95} INFO - [2019-02-21 16:15:01,335] {base_hook.py:83} INFO - Using connection to: id: sendgrid_templates_test. Host: https://api.sendgrid.com/, Port: None, Schema: None, Login: None, Password: XXXXXXXX, extra: {'authorization': 'Bearer [my token]'}
[2019-02-21 16:15:01,338] {logging_mixin.py:95} INFO - [2019-02-21 16:15:01,337] {http_hook.py:126} INFO - Sending 'GET' to url: https://api.sendgrid.com//v3/templates
[2019-02-21 16:15:01,956] {logging_mixin.py:95} WARNING - /home/csconnell/.pyenv/versions/airflow/lib/python3.6/site-packages/urllib3/connectionpool.py:847: InsecureRequestWarning: Unverified HTTPS request is being made. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/latest/advanced-usage.html#ssl-warnings
InsecureRequestWarning)
[2019-02-21 16:15:05,242] {logging_mixin.py:95} INFO - [2019-02-21 16:15:05,241] {jobs.py:2527} INFO - Task exited with return code 0
AIRFLOW_CONN_HTTP_EXAMPLE=http://https%3a%2f%2fexample.com/
而不是:
AIRFLOW_CONN_HTTP_EXAMPLE=https://example.com/
或在 UI 上设置连接。我们可以将以下一行代码放入HttpsOperator(SimpleHttpOperator)的上方,而不是实现HttpsHook:
...
self.extra_options['verify'] = True
response = http.run(self.endpoint,
self.data,
self.headers,
self.extra_options)
...
在Airflow 2中,该问题已经得到解决。 只需检查以下内容:
主机
名不要以/
结尾SimpleHttpOperator
的endpoint
参数以/
开头
curl
了吗? - tobi6