datax-admin/app/core/airflow/uri.py

import json

import requests

from app.common.minio import FileHandler


def spark_result_tb_name(job_id,task_id, spark_node_id, out_pin, is_tmp=False):
    return f'job{job_id}_task{task_id}_subnode{spark_node_id}_output{out_pin}{"_tmp" if is_tmp else ""}'


def get_sub_task_script_uri(task_id, sub_node_id):
    return f'/xxx/task_{task_id}/sub_{sub_node_id}.py'


def get_remote_file(uri, return_json=False):
    data = requests.get(uri).content
    return json.loads(data) if return_json else data


def upload2oss(content: bytes, uri: str, minio_bucket: str):
    minio_handler = FileHandler(bucket_name=minio_bucket)
    minio_handler.put_byte_file(file_name=uri, file_content=content)


def get_job_path(job_id):
    dag_path = '/dags/'
    return dag_path + f'dag_{job_id}.py'


def get_airflow_api_info():
    uri_prefix = f'http://192.168.199.109/api/v1'
    headers = {
        'content-type': 'application/json',
        'Authorization': 'basic YWRtaW46YWRtaW4=',
        'Host': 'airflow-web.sxkj.com'
    }
    return uri_prefix, headers