stroblmeandClaude Opus 5
cba8ca2c54
Publishing or deleting a flow now splices that one flow into the running graph instead of reconnecting every node in the installation, saving a shared node's source rebuilds the flows using it, and installing modules rebuilds only the flows holding a node that would not load. Renaming stays on the full rebuild — it rewrites message references in every other flow's document — and so does startup, which has no graph to splice into. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""The python packages node code may import.
|
|
|
|
A manifest in the flow store, a venv on the data volume, and one button that
|
|
brings the second in line with the first. Nothing here restarts the engine:
|
|
the worker pool retires its processes, and the next node call picks up the new
|
|
packages.
|
|
"""
|
|
|
|
import time
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from fastapi.concurrency import run_in_threadpool
|
|
|
|
from fluksio.api.deps import (
|
|
CurrentUser,
|
|
FlowControllerDep,
|
|
WorkerPoolDep,
|
|
get_current_user,
|
|
)
|
|
from fluksio.flow import modules
|
|
from fluksio.flow.events import event_bus
|
|
from fluksio.flow.schemas import ApplyRequest, ApplyResult, ModulesInfo
|
|
|
|
router = APIRouter(
|
|
prefix="/modules", tags=["modules"], dependencies=[Depends(get_current_user)]
|
|
)
|
|
|
|
|
|
@router.get("/", response_model=ModulesInfo)
|
|
async def read_modules(controller: FlowControllerDep) -> Any:
|
|
"""What node code can import, and whether it matches the manifest."""
|
|
return await run_in_threadpool(modules.info, controller.store)
|
|
|
|
|
|
@router.post("/apply", response_model=ApplyResult)
|
|
async def apply_modules(
|
|
body: ApplyRequest,
|
|
controller: FlowControllerDep,
|
|
pool: WorkerPoolDep,
|
|
user: CurrentUser,
|
|
) -> Any:
|
|
"""Install exactly these requirements, then hand them to the workers.
|
|
|
|
A manifest that does not resolve changes nothing: the venv is left as it
|
|
was and the stored manifest is only written once the install succeeded.
|
|
|
|
Only the flows already holding a node that would not load are rebuilt,
|
|
because those are the ones an install is called to fix. A flow that this
|
|
install *breaks* — a package taken back out from under it — is still green
|
|
and fails at call time with the node author's own import error, until
|
|
something rebuilds it.
|
|
"""
|
|
ok, output = await run_in_threadpool(modules.sync, body.requirements)
|
|
if not ok:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=output or "These requirements could not be installed",
|
|
)
|
|
await run_in_threadpool(controller.store.write_requirements, body.requirements)
|
|
event_bus.publish(
|
|
{
|
|
"type": "audit",
|
|
"action": "installed modules",
|
|
"flow": "",
|
|
"user": user.email,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
# Retire the workers first, so the rebuild compiles every node against the
|
|
# packages that were just installed — a node that could not import one is
|
|
# the reason this was called, and it stays red until it is built again.
|
|
pool.respawn_all()
|
|
await controller.reload_failed_flows()
|
|
return ApplyResult(ok=True, output=output)
|