1. # coding: utf-8
    2. import os
    3. import time
    4. import json
    5. import subprocess
    6. import pandas as pd
    7. import streamlit as st
    10. st.set_page_config(
    11.      page_title="Airflow Deployment Platform",
    12.      page_icon="🚀",
    13.      layout="wide",
    14.      initial_sidebar_state="expanded",
    15.     )
    17. st.markdown("""
    18. <style>
    19. .big-font {
    20.     font-size:22px
    21. }
    22. </style>
    23. """, unsafe_allow_html=True)
    25. @st.cache(show_spinner=False, suppress_st_warning=True, ttl=120)
    26. def load_airflow_dags():
    27.     dags_json = os.popen("airflow dags list -o json")
    28.     data = dags_json.readlines()[1]
    29.     return data
    32. with st.sidebar:
    33.     st.title("📊 部署和更新Airflow任务")
    34.     option = st.selectbox("任务方式", ('更新', '部署'))
    36. airflow_home = os.environ.get("AIRFLOW_HOME")
    38. if option == "更新":
    39.     st.title("更新Airflow任务")
    40.     st.markdown("[🚀访问Airflow ](http://airflow.com:9000/)")
    42.     with st.spinner("🔎 后台查询任务中，请稍等..."):
    43.         dags_json = os.popen("airflow dags list -o json")
    44.         data = load_airflow_dags()
    45.         df = pd.DataFrame.from_dict(json.loads(data))
    46.         df = df.loc[df["owner"].isin(["link", "yumingmin", "mock"])].reset_index(drop=True)
    48.     dag_id, owner, update_col = st.columns(3)
    50.     dag_id.markdown("**DAG_ID**")
    51.     owner.markdown("**OWNER**")
    52.     update_col.markdown("**UPDATE**")
    54.     for raw in df.values:
    55.         project_path = os.path.join(airflow_home, "dags", raw[1].split("/")[0])
    57.         dag_id.markdown('<p class="big-font">%s</p>' % raw[0], unsafe_allow_html=True)
    58.         owner.markdown('<p class="big-font">%s</p>' % raw[2], unsafe_allow_html=True)
    60.         if update_col.button("更新", key=raw[0]):
    61.             update_cmd = subprocess.Popen(f"cd {project_path} && git pull", shell=True, 
    62.                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    64.             stdout, stderr  = update_cmd.communicate()
    65.             update_cmd.wait()
    67.             succ_flag = 1 if 'up to date' in str(stdout) or 'done' in str(stdout) else 0
    69.             if succ_flag == 1:
    70.                 with st.spinner("😁更新成功!!"):
    71.                     time.sleep(5)
    72.             else:
    73.                 with st.spinner("😂更新失败!!"):
    74.                     time.sleep(5)
    76. if option == "部署":
    77.     st.title("部署Airflow任务")
    78.     st.markdown("[🚀访问Airflow ](http://airflow.com:9000/)")
    79.     form = st.form(key="Deploy-Project")
    80.     gitlab_url = form.text_input('输入 GitLab 项目地址: 例如 git@github.com:yumingmin/testisok.git')
    81.     submit = form.form_submit_button("部署")
    83.     if submit:
    84.         if gitlab_url:
    85.             with st.spinner("⏳ 后台部署中，请稍等..."):
    86.                 deploy_cmd = subprocess.Popen(f"cd {airflow_home}/dags && git clone {gitlab_url}", 
    87.                                         shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    88.                 stdout, stderr  = deploy_cmd.communicate()
    89.                 deploy_cmd.wait()
    90.                 succ_flag = 1 if 'Cloning' in str(stderr) else 0
    92.                 st.text("out %s" % str(stderr))
    93.                 st.text("err %s" % str(stderr))
    95.                 if succ_flag == 1:
    96.                     with st.spinner("😁部署成功!!"):
    97.                         time.sleep(5)
    98.                 else:
    99.                     with st.spinner("😂部署失败!!"):
    100.                         time.sleep(5)
    101.         else:
    102.             with st.spinner("🚧 必须填写项目地址！！！"):
    103.                 time.sleep(5)

    终端执行：

    1. $ nohup streamlit run update_gitlab_projects_dashboard.py --server.port 9001 > streamlit_run.log &