core/engine/src/services/flow_service.py - main - alexefan136/flowstack

core/engine/src/services/flow_service.py
133 строки5 KB
Alexander Efanov
upd fix
31 июл 2026, 19:17
31 июл 2026, 19:17d146d86
"""Бизнес-логика flows (CRUD + запуск через GraphExecutor)."""

from __future__ import annotations

import uuid
from collections.abc import AsyncIterator
from typing import Any

import structlog
from sqlalchemy.ext.asyncio import AsyncSession

from src.db.repositories import FlowRepository
from src.primitives import Flow, FlowEdge, FlowEdgeType, FlowNode, FlowNodeType, Task
from src.runtime import GraphExecutor, create_graph_executor
from src.schemas import FlowCreate, FlowUpdate

logger = structlog.get_logger()


class FlowService:
    """Сервис для работы с flows."""

    def __init__(
        self,
        session: AsyncSession,
        user_id: uuid.UUID,
        workspace_id: str,
        executor: GraphExecutor | None = None,
    ) -> None:
        self.session = session
        self.user_id = user_id
        self.workspace_id = workspace_id
        self.flow_repo = FlowRepository(session, user_id, workspace_id)
        self.executor = executor or create_graph_executor()

    # ========================================================================
    # CRUD
    # ========================================================================

    async def create(self, data: FlowCreate) -> Any:
        """Создать flow."""
        flow = await self.flow_repo.create(
            user_id=self.user_id,
            workspace_id=self.workspace_id,
            name=data.name,
            description=data.description,
            nodes=[n.model_dump() for n in data.nodes],
            edges=[e.model_dump() for e in data.edges],
            status=data.status,
        )
        logger.info("flow.created", flow_id=str(flow.id), name=data.name)
        return flow

    async def get(self, flow_id: uuid.UUID) -> Any | None:
        return await self.flow_repo.get(flow_id)

    async def list(self, status: str | None = None, limit: int = 50, offset: int = 0) -> list[Any]:
        return await self.flow_repo.list(status=status, limit=limit, offset=offset)

    async def update(self, flow_id: uuid.UUID, data: FlowUpdate) -> Any | None:
        updates = data.model_dump(exclude_unset=True)
        if not updates:
            return await self.flow_repo.get(flow_id)
        return await self.flow_repo.update(flow_id, **updates)

    async def delete(self, flow_id: uuid.UUID) -> bool:
        return await self.flow_repo.delete(flow_id)

    # ========================================================================
    # Conversion
    # ========================================================================

    def _to_flow(self, flow_db: Any) -> Flow:
        """Сконвертировать DB-модель flow в primitives.Flow."""
        nodes = [
            FlowNode(
                id=n["id"],
                type=FlowNodeType(n["type"]),
                label=n.get("label", ""),
                data=n.get("data", {}),
                position=n.get("position", {}),
                config=n.get("config", {}),
            )
            for n in (flow_db.nodes or [])
        ]
        edges = [
            FlowEdge(
                id=e["id"],
                source=e["source"],
                target=e["target"],
                label=e.get("label", ""),
                edge_type=FlowEdgeType(e.get("edge_type", "normal")),
                condition=e.get("condition"),
                priority=e.get("priority", 0),
            )
            for e in (flow_db.edges or [])
        ]
        return Flow(
            id=str(flow_db.id),
            name=flow_db.name,
            description=flow_db.description or "",
            nodes=nodes,
            edges=edges,
        )

    # ========================================================================
    # Run
    # ========================================================================

    async def run(self, flow_id: uuid.UUID, input_data: dict[str, Any]) -> Task:
        """Запустить flow (non-streaming)."""
        flow_db = await self.flow_repo.get(flow_id)
        if flow_db is None:
            raise ValueError(f"Flow {flow_id} not found")

        flow = self._to_flow(flow_db)
        logger.info("flow.run.start", flow_id=str(flow_id))
        task = await self.executor.run(flow, input_data)
        logger.info("flow.run.done", flow_id=str(flow_id), status=task.status.value)
        return task

    async def run_stream(
        self, flow_id: uuid.UUID, input_data: dict[str, Any]
    ) -> AsyncIterator[dict[str, Any]]:
        """Запустить flow со стримингом."""
        flow_db = await self.flow_repo.get(flow_id)
        if flow_db is None:
            yield {"type": "error", "error": f"Flow {flow_id} not found"}
            return

        flow = self._to_flow(flow_db)
        async for event in self.executor.run_stream(flow, input_data):
            yield event