Наше приложение
import streamlit as st
from kubernetes import client, config
from kubernetes.client.rest import ApiException
# Для взаимодействия с кластером Kubernetes нам нужен конфиг,
# получаем его так
def load_kube_config():
try:
config.load_incluster_config()
except config.ConfigException:
config.load_kube_config()
# Генерация манифеста для K8S.
# В проде, конечно, возможно задать больше функций
def build_application_set(name: str) -> dict:
return {
"apiVersion": "argoproj.io/v1alpha1",
"kind": "ApplicationSet",
"metadata": {
"name": f"ministand-{name}",
"namespace": "argo-cd",
},
"spec": {
"template": {
"metadata": {
"name": f"{name}-app",
},
"spec": {
"project": "default",
"destination": {
"server": "https://kubernetes.default.svc",
"namespace": f"ministand-{name}",
},
},
},
},
}
# Деплой стенда через K8S API
def deploy_stand(name: str):
api = client.CustomObjectsApi()
manifest = build_application_set(name)
try:
# применение манифеста в k8s, обращение к API
api.create_namespaced_custom_object(
group="argoproj.io",
version="v1alpha1",
namespace="argo-cd",
plural="applicationsets",
body=manifest,
)
st.success(f"✅ Стенд {name} создан")
st.rerun()
except ApiException as e:
if e.status == 409:
st.error(f"❗ Стенд {name} уже существует")
else:
st.error(f"Ошибка: {e}")
# Получение списка стендов в кластере K8S
def get_stands():
api = client.CustomObjectsApi()
try:
resp = api.list_namespaced_custom_object(
group="argoproj.io",
version="v1alpha1",
namespace="argo-cd",
plural="applicationsets",
)
return resp.get("items", [])
except ApiException as e:
st.error(f"Ошибка Kubernetes API: {e}")
return []
# Самое главное - UI
st.set_page_config(page_title="Stand Manager", layout="wide")
st.title("🛠️ Менеджер стендов")
# Инициализация один раз
if "kube_loaded" not in st.session_state:
load_kube_config()
st.session_state["kube_loaded"] = True
# Тут красиво оформляем список стендов,
# примеры того как это выглядит на проде будут ниже
stands = get_stands()
for stand in stands:
st.subheader(stand["metadata"]["name"])
col1, col2, col3 = st.columns([4, 1, 1])
# Название
with col1:
st.subheader(stand["metadata"]["name"])
# Открыть
with col2:
if st.button("Открыть", key=stand["metadata"]["uid"]):
# Тут должна быть функция раскрытия подробностей
# Удалить
with col3:
if st.button("Удалить", key="del-" + stand["metadata"]["uid"]):
# Тут должна быть функция удаления
st.divider()
# Форма создания нового стенда
st.subheader("🚀 Новый стенд")
with st.form("deploy_form"):
name = st.text_input("Имя стенда")
if st.form_submit_button("Создать"):
deploy_stand(name)
