datax-admin/app/utils/send_util.py

from unittest import result
import requests
from configs.settings import config

HOST = config.get('AIRFLOW', 'HOST')
PORT = config.get('AIRFLOW', 'PORT')

def send_post(uri,data):
    res = requests.post(url=f'http://{HOST}:{PORT}{uri}', json=data)
    result = res.json()
    if 'code' in result.keys() and result['code'] == 200:
        return res.json()
    else:
        raise Exception('请求airflow失败')

def send_submit(af_job_id):
    res = requests.post(url=f'http://{HOST}:{PORT}/jpt/af_job/submit?id='+str(af_job_id))
    result = res.json()
    print(result)
    if 'code' in result.keys() and result['code'] == 200:
        return res.json()
    else:
        raise Exception('请求airflow失败')


def send_put(uri,path_data,data):
    res = requests.put(url=f'http://{HOST}:{PORT}{uri}/{path_data}', json=data)
    result = res.json()
    if 'code' in result.keys() and result['code'] == 200:
        return res.json()
    else:
        raise Exception('请求airflow失败')

def send_get(uri,path_data):
    res = requests.get(url=f'http://{HOST}:{PORT}{uri}/{path_data}')
    result = res.json()
    if 'code' in result.keys() and result['code'] == 200:
        return res.json()
    else:
        raise Exception('请求airflow失败')