1fbf9185a2
- Добавлена колонка 'Тип' во все таблицы истории сборок - Для push операций отображается registry вместо платформ - Сохранение пользователя при создании push лога - Исправлена ошибка с logger в push_docker_image endpoint - Улучшено отображение истории сборок с визуальными индикаторами
313 lines
11 KiB
Python
313 lines
11 KiB
Python
"""
|
||
API endpoints для управления ролями
|
||
Автор: Сергей Антропов
|
||
Сайт: https://devops.org.ru
|
||
"""
|
||
|
||
from fastapi import APIRouter, Request, HTTPException, Form
|
||
from fastapi.responses import HTMLResponse, JSONResponse
|
||
from fastapi.templating import Jinja2Templates
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional
|
||
import yaml
|
||
import json
|
||
from app.core.config import settings
|
||
from app.services.role_service import RoleService
|
||
|
||
router = APIRouter()
|
||
templates_path = Path(__file__).parent.parent.parent.parent / "templates"
|
||
templates = Jinja2Templates(directory=str(templates_path))
|
||
role_service = RoleService()
|
||
|
||
|
||
def get_roles_list() -> List[Dict]:
|
||
"""Получение списка ролей"""
|
||
roles_dir = settings.PROJECT_ROOT / "roles"
|
||
|
||
if not roles_dir.exists():
|
||
return []
|
||
|
||
roles = []
|
||
for role_dir in roles_dir.iterdir():
|
||
if role_dir.is_dir():
|
||
role_info = {
|
||
"name": role_dir.name,
|
||
"path": str(role_dir),
|
||
"has_tasks": (role_dir / "tasks" / "main.yml").exists(),
|
||
"has_defaults": (role_dir / "defaults" / "main.yml").exists(),
|
||
"has_handlers": (role_dir / "handlers" / "main.yml").exists(),
|
||
"has_meta": (role_dir / "meta" / "main.yml").exists(),
|
||
"has_readme": (role_dir / "README.md").exists(),
|
||
"description": ""
|
||
}
|
||
|
||
# Чтение описания из meta/main.yml
|
||
meta_file = role_dir / "meta" / "main.yml"
|
||
if meta_file.exists():
|
||
try:
|
||
with open(meta_file) as f:
|
||
meta_data = yaml.safe_load(f)
|
||
if meta_data and isinstance(meta_data, dict):
|
||
role_info["description"] = meta_data.get("galaxy_info", {}).get("description", "")
|
||
role_info["author"] = meta_data.get("galaxy_info", {}).get("author", "")
|
||
role_info["platforms"] = meta_data.get("galaxy_info", {}).get("platforms", [])
|
||
except:
|
||
pass
|
||
|
||
roles.append(role_info)
|
||
|
||
return sorted(roles, key=lambda x: x["name"])
|
||
|
||
|
||
@router.get("/roles/create", response_class=HTMLResponse)
|
||
async def create_role_page(request: Request):
|
||
"""Страница создания роли"""
|
||
return templates.TemplateResponse(
|
||
"pages/roles/create.html",
|
||
{"request": request}
|
||
)
|
||
|
||
|
||
@router.get("/roles/{role_name}/edit", response_class=HTMLResponse)
|
||
async def edit_role_page(request: Request, role_name: str):
|
||
"""Страница редактирования роли"""
|
||
roles = get_roles_list()
|
||
role = next((r for r in roles if r["name"] == role_name), None)
|
||
|
||
if not role:
|
||
raise HTTPException(status_code=404, detail=f"Роль '{role_name}' не найдена")
|
||
|
||
role_dir = settings.PROJECT_ROOT / "roles" / role_name
|
||
|
||
# Чтение всех файлов роли
|
||
files_content = {}
|
||
for file_type in ["tasks", "handlers", "defaults", "vars", "meta"]:
|
||
file_path = role_dir / file_type / "main.yml"
|
||
if file_path.exists():
|
||
files_content[file_type] = file_path.read_text()
|
||
else:
|
||
files_content[file_type] = ""
|
||
|
||
readme_content = ""
|
||
readme_file = role_dir / "README.md"
|
||
if readme_file.exists():
|
||
readme_content = readme_file.read_text()
|
||
else:
|
||
readme_content = ""
|
||
|
||
return templates.TemplateResponse(
|
||
"pages/roles/edit.html",
|
||
{
|
||
"request": request,
|
||
"role": role,
|
||
"files_content": files_content,
|
||
"readme_content": readme_content
|
||
}
|
||
)
|
||
|
||
|
||
@router.get("/roles", response_class=HTMLResponse)
|
||
async def list_roles(
|
||
request: Request,
|
||
page: int = 1,
|
||
per_page: int = 10,
|
||
search: Optional[str] = None
|
||
):
|
||
"""Страница списка ролей с пагинацией"""
|
||
"""Страница списка ролей с пагинацией"""
|
||
roles = get_roles_list()
|
||
|
||
# Фильтрация по поиску
|
||
if search:
|
||
search_lower = search.lower()
|
||
roles = [
|
||
r for r in roles
|
||
if search_lower in r["name"].lower() or
|
||
search_lower in (r.get("description", "") or "").lower()
|
||
]
|
||
|
||
# Пагинация
|
||
total = len(roles)
|
||
total_pages = (total + per_page - 1) // per_page if total > 0 else 1
|
||
page = max(1, min(page, total_pages))
|
||
start = (page - 1) * per_page
|
||
end = start + per_page
|
||
paginated_roles = roles[start:end]
|
||
|
||
return templates.TemplateResponse(
|
||
"pages/roles/list.html",
|
||
{
|
||
"request": request,
|
||
"roles": paginated_roles,
|
||
"total": total,
|
||
"page": page,
|
||
"per_page": per_page,
|
||
"total_pages": total_pages,
|
||
"search": search or ""
|
||
}
|
||
)
|
||
|
||
|
||
@router.get("/api/v1/roles")
|
||
async def get_roles_api():
|
||
"""API endpoint для получения списка ролей"""
|
||
return get_roles_list()
|
||
|
||
|
||
@router.get("/api/v1/roles/{role_name}")
|
||
async def get_role_info(role_name: str):
|
||
"""API endpoint для получения информации о роли"""
|
||
roles = get_roles_list()
|
||
role = next((r for r in roles if r["name"] == role_name), None)
|
||
if not role:
|
||
raise HTTPException(status_code=404, detail=f"Роль '{role_name}' не найдена")
|
||
return role
|
||
|
||
|
||
@router.get("/roles/{role_name}", response_class=HTMLResponse)
|
||
async def role_detail(request: Request, role_name: str):
|
||
"""Страница деталей роли"""
|
||
roles = get_roles_list()
|
||
role = next((r for r in roles if r["name"] == role_name), None)
|
||
|
||
if not role:
|
||
raise HTTPException(status_code=404, detail=f"Роль '{role_name}' не найдена")
|
||
|
||
# Чтение содержимого файлов
|
||
role_dir = settings.PROJECT_ROOT / "roles" / role_name
|
||
|
||
tasks_content = ""
|
||
tasks_file = role_dir / "tasks" / "main.yml"
|
||
if tasks_file.exists():
|
||
tasks_content = tasks_file.read_text()
|
||
|
||
defaults_content = ""
|
||
defaults_file = role_dir / "defaults" / "main.yml"
|
||
if defaults_file.exists():
|
||
defaults_content = defaults_file.read_text()
|
||
|
||
readme_content = ""
|
||
readme_file = role_dir / "README.md"
|
||
if readme_file.exists():
|
||
readme_content = readme_file.read_text()
|
||
|
||
return templates.TemplateResponse(
|
||
"pages/roles/detail.html",
|
||
{
|
||
"request": request,
|
||
"role": role,
|
||
"tasks_content": tasks_content,
|
||
"defaults_content": defaults_content,
|
||
"readme_content": readme_content
|
||
}
|
||
)
|
||
|
||
|
||
@router.post("/api/v1/roles/create")
|
||
async def create_role_api(
|
||
role_name: str = Form(...),
|
||
template: str = Form("default"),
|
||
description: str = Form(""),
|
||
platforms: str = Form(""), # JSON строка или через запятую
|
||
variables: str = Form("") # JSON строка
|
||
):
|
||
"""API endpoint для создания роли"""
|
||
try:
|
||
# Парсинг platforms
|
||
platforms_list = []
|
||
if platforms:
|
||
if platforms.startswith("["):
|
||
platforms_list = json.loads(platforms)
|
||
else:
|
||
platforms_list = [p.strip() for p in platforms.split(",") if p.strip()]
|
||
|
||
# Парсинг variables
|
||
variables_list = []
|
||
if variables:
|
||
variables_list = json.loads(variables)
|
||
|
||
result = role_service.create_role(
|
||
role_name=role_name,
|
||
template=template,
|
||
description=description,
|
||
platforms=platforms_list,
|
||
variables=variables_list
|
||
)
|
||
|
||
return JSONResponse(content=result, status_code=201)
|
||
except ValueError as e:
|
||
raise HTTPException(status_code=400, detail=str(e))
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"Ошибка при создании роли: {str(e)}")
|
||
|
||
|
||
@router.post("/api/v1/roles/{role_name}/update")
|
||
async def update_role_api(
|
||
role_name: str,
|
||
file_type: str = Form(...),
|
||
content: str = Form(...)
|
||
):
|
||
"""API endpoint для обновления файла роли"""
|
||
try:
|
||
role_dir = settings.PROJECT_ROOT / "roles" / role_name
|
||
|
||
if not role_dir.exists():
|
||
raise HTTPException(status_code=404, detail=f"Роль '{role_name}' не найдена")
|
||
|
||
# Определение пути к файлу
|
||
if file_type == "readme":
|
||
file_path = role_dir / "README.md"
|
||
elif file_type in ["tasks", "handlers", "defaults", "vars", "meta"]:
|
||
file_path = role_dir / file_type / "main.yml"
|
||
file_path.parent.mkdir(parents=True, exist_ok=True)
|
||
else:
|
||
raise HTTPException(status_code=400, detail=f"Неверный тип файла: {file_type}")
|
||
|
||
# Сохранение файла
|
||
file_path.write_text(content)
|
||
|
||
return JSONResponse(content={
|
||
"success": True,
|
||
"message": f"Файл {file_type} успешно обновлен"
|
||
})
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"Ошибка при обновлении роли: {str(e)}")
|
||
|
||
|
||
@router.get("/roles/{role_name}/deploy", response_class=HTMLResponse)
|
||
async def deploy_role_page(request: Request, role_name: str):
|
||
"""Страница деплоя роли"""
|
||
roles = get_roles_list()
|
||
role = next((r for r in roles if r["name"] == role_name), None)
|
||
|
||
if not role:
|
||
raise HTTPException(status_code=404, detail=f"Роль '{role_name}' не найдена")
|
||
|
||
# Проверка наличия inventory
|
||
inventory_file = settings.PROJECT_ROOT / "inventory" / "hosts.ini"
|
||
inventory_exists = inventory_file.exists()
|
||
|
||
# Чтение inventory если существует
|
||
inventory_content = ""
|
||
if inventory_exists:
|
||
try:
|
||
inventory_content = inventory_file.read_text()
|
||
except Exception:
|
||
pass
|
||
|
||
# Проверка наличия deploy.yml
|
||
deploy_playbook = settings.PROJECT_ROOT / "roles" / "deploy.yml"
|
||
deploy_playbook_exists = deploy_playbook.exists()
|
||
|
||
return templates.TemplateResponse(
|
||
"pages/roles/deploy.html",
|
||
{
|
||
"request": request,
|
||
"role": role,
|
||
"role_name": role_name,
|
||
"inventory_exists": inventory_exists,
|
||
"inventory_content": inventory_content,
|
||
"deploy_playbook_exists": deploy_playbook_exists
|
||
}
|
||
)
|