backend/main.py - master - PashigorevAY/TestFastAPI

backend/main.py
303 строки8 KB
Pashigorev
Initial commit
05 июл 2026, 21:43
05 июл 2026, 21:4319770a6
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, field_validator
from typing import List, Dict, Any, Optional
import yaml
import re
from pathlib import Path

app = FastAPI(title="YAML Editor API")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

BASE_DIR = Path(__file__).parent.parent
RESOURCES_DIR = BASE_DIR / "resources"
DIRECTORIES_DIR = RESOURCES_DIR / "directories"
TASKS_DIR = RESOURCES_DIR / "tasks"
PATHS_DIR = RESOURCES_DIR / "paths"


class Person(BaseModel):
    code: str
    name: str
    position: str


class PeopleData(BaseModel):
    people: List[Person]


class StatusesData(BaseModel):
    statuses: List[str]


class Task(BaseModel):
    id: str
    project: str
    title: str
    description: str
    assignee: str
    responsibles: List[str]
    due_date: str
    labels: List[str]
    assignee_to_epic_reporter: bool


class CatalogInfo(BaseModel):
    name: str
    description: str


class CreateCatalogRequest(BaseModel):
    folder_name: str
    description: str

    @field_validator('folder_name')
    @classmethod
    def validate_folder_name(cls, v: str) -> str:
        if not re.match(r'^[a-z0-9_]+$', v):
            raise ValueError('Имя папки должно содержать только маленькие английские буквы, цифры и подчеркивания')
        return v


def load_yaml(file_path: Path) -> Dict[str, Any]:
    if not file_path.exists():
        raise HTTPException(status_code=404, detail=f"File not found: {file_path}")
    with open(file_path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


def save_yaml(file_path: Path, data: Dict[str, Any]) -> None:
    file_path.parent.mkdir(parents=True, exist_ok=True)
    with open(file_path, "w", encoding="utf-8") as f:
        yaml.dump(data, f, allow_unicode=True, sort_keys=False)


def load_template() -> Dict[str, Any]:
    template_path = TASKS_DIR / "template.yaml"
    return load_yaml(template_path)


@app.get("/api/people")
def get_people():
    file_path = DIRECTORIES_DIR / "people.yaml"
    data = load_yaml(file_path)
    return data


@app.put("/api/people")
def save_people(data: PeopleData):
    file_path = DIRECTORIES_DIR / "people.yaml"
    save_yaml(file_path, data.model_dump())
    return {"status": "success"}


@app.get("/api/statuses")
def get_statuses():
    file_path = DIRECTORIES_DIR / "statuses.yaml"
    data = load_yaml(file_path)
    return data


@app.put("/api/statuses")
def save_statuses(data: StatusesData):
    file_path = DIRECTORIES_DIR / "statuses.yaml"
    save_yaml(file_path, data.model_dump())
    return {"status": "success"}


@app.get("/api/tasks/catalogs")
def get_catalogs():
    catalogs = []
    if not TASKS_DIR.exists():
        return {"catalogs": []}
    
    for item in TASKS_DIR.iterdir():
        if item.is_dir() and item.name != "template.yaml":
            readme_path = item / "README.md"
            description = ""
            if readme_path.exists():
                description = readme_path.read_text(encoding="utf-8")
            catalogs.append({
                "name": item.name,
                "description": description
            })
    return {"catalogs": catalogs}


@app.post("/api/tasks/catalogs")
def create_catalog(request: CreateCatalogRequest):
    catalog_path = TASKS_DIR / request.folder_name
    if catalog_path.exists():
        raise HTTPException(status_code=400, detail="Каталог уже существует")
    
    catalog_path.mkdir(parents=True, exist_ok=True)
    readme_path = catalog_path / "README.md"
    readme_path.write_text(request.description, encoding="utf-8")
    
    return {"status": "success", "name": request.folder_name}


@app.get("/api/tasks/catalogs/{catalog_name}/tasks")
def get_tasks(catalog_name: str):
    catalog_path = TASKS_DIR / catalog_name
    if not catalog_path.exists():
        raise HTTPException(status_code=404, detail="Каталог не найден")
    
    tasks = []
    for item in catalog_path.glob("task*.yaml"):
        if item.is_file():
            try:
                data = load_yaml(item)
                tasks.append(data)
            except Exception:
                pass
    
    tasks.sort(key=lambda x: x.get("id", ""))
    return {"tasks": tasks}


@app.get("/api/tasks/catalogs/{catalog_name}/tasks/{task_id}")
def get_task(catalog_name: str, task_id: str):
    catalog_path = TASKS_DIR / catalog_name
    task_path = catalog_path / f"{task_id}.yaml"
    if not task_path.exists():
        raise HTTPException(status_code=404, detail="Задача не найдена")
    return load_yaml(task_path)


@app.post("/api/tasks/catalogs/{catalog_name}/tasks")
def create_task(catalog_name: str):
    catalog_path = TASKS_DIR / catalog_name
    if not catalog_path.exists():
        raise HTTPException(status_code=404, detail="Каталог не найден")
    
    template = load_template()
    
    task_files = list(catalog_path.glob("task*.yaml"))
    max_num = 0
    for f in task_files:
        match = re.match(r"task(\d+)\.yaml", f.name)
        if match:
            num = int(match.group(1))
            if num > max_num:
                max_num = num
    
    task_num = max_num + 1
    task_id = f"task{task_num}"
    
    new_task = template.copy()
    new_task["id"] = task_id
    
    task_path = catalog_path / f"{task_id}.yaml"
    save_yaml(task_path, new_task)
    
    return new_task


@app.put("/api/tasks/catalogs/{catalog_name}/tasks/{task_id}")
def save_task(catalog_name: str, task_id: str, task: Task):
    catalog_path = TASKS_DIR / catalog_name
    if not catalog_path.exists():
        raise HTTPException(status_code=404, detail="Каталог не найден")
    
    task_path = catalog_path / f"{task_id}.yaml"
    save_yaml(task_path, task.model_dump())
    return {"status": "success"}


class PathStep(BaseModel):
    id: str
    name: str
    optional: bool
    statuses: list[str]
    depends_on: list[str]


class PathItem(BaseModel):
    tasks_group: str
    path: list[PathStep]


class PathsData(BaseModel):
    paths: dict[str, PathItem]


class CreatePathRequest(BaseModel):
    name: str
    tasks_group: str


@app.get("/api/paths")
def get_paths():
    paths_file = PATHS_DIR / "paths.yaml"
    if not paths_file.exists():
        return {"paths": {}}
    return load_yaml(paths_file)


@app.post("/api/paths")
def create_path(request: CreatePathRequest):
    paths_file = PATHS_DIR / "paths.yaml"
    data = load_yaml(paths_file) if paths_file.exists() else {"paths": {}}
    
    if request.name in data["paths"]:
        raise HTTPException(status_code=400, detail="Путь с таким именем уже существует")
    
    data["paths"][request.name] = {
        "tasks_group": request.tasks_group,
        "path": []
    }
    
    save_yaml(paths_file, data)
    return {"status": "success"}


@app.put("/api/paths/{path_name}")
def save_path(path_name: str, path_item: PathItem):
    paths_file = PATHS_DIR / "paths.yaml"
    data = load_yaml(paths_file)
    
    if path_name not in data["paths"]:
        raise HTTPException(status_code=404, detail="Путь не найден")
    
    data["paths"][path_name] = path_item.model_dump()
    save_yaml(paths_file, data)
    return {"status": "success"}


@app.delete("/api/paths/{path_name}")
def delete_path(path_name: str):
    paths_file = PATHS_DIR / "paths.yaml"
    data = load_yaml(paths_file)
    
    if path_name not in data["paths"]:
        raise HTTPException(status_code=404, detail="Путь не найден")
    
    del data["paths"][path_name]
    save_yaml(paths_file, data)
    return {"status": "success"}


@app.get("/api/tasks-groups")
def get_tasks_groups():
    groups = []
    if not TASKS_DIR.exists():
        return {"groups": []}
    
    for item in TASKS_DIR.iterdir():
        if item.is_dir() and item.name != "template.yaml":
            groups.append(item.name)
    
    return {"groups": groups}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)