"""Build routes on /openapi/v1 for workflow and advanced-chat apps: run history, publishing and versions, and draft environment variables.""" from __future__ import annotations from collections.abc import Iterable from http import HTTPStatus from typing import Final from uuid import UUID from flask_restx import Resource from sqlalchemy.orm import sessionmaker from werkzeug.exceptions import BadRequest from constants import HIDDEN_VALUE from constants.oauth_bearer import Scope from controllers.common.fields import SimpleResultResponse from controllers.common.rbac import RBACPermission from controllers.openapi import openapi_ns from controllers.openapi._contract import Example, Kind, endpoint from controllers.openapi._errors import ( DraftNotFound, OpenApiError, RunNotFound, SecretMaskNotSecret, SecretMaskUnknownId, VersionNotFound, VersionNotRestorable, ) from controllers.openapi._models import ( EnvVariableListResponse, EnvVariableRow, EnvVariableSetPayload, EnvVariableValueType, Hint, PublishPayload, PublishResponse, RestoreResponse, RunListQuery, RunListResponse, VersionListQuery, VersionListResponse, VersionRow, ) from controllers.openapi.app_run import require_mode from controllers.openapi.auth.context import Context from controllers.openapi.auth.requirements import account_app_guards from core.helper import encrypter from core.logging.context import get_request_id, get_trace_id from core.workflow.llm_environment_variable import environment_variable_value_type from extensions.ext_application_services import application_services from extensions.ext_database import db from factories import variable_factory from fields.workflow_run_fields import ( WorkflowRunDetailResponse, WorkflowRunNodeExecutionListResponse, workflow_run_pagination_response_source, workflow_run_response_source, ) from graphon.variables import SecretVariable, VariableBase from graphon.variables.exc import VariableError from libs.helper import to_timestamp from machinery.context import RequestContext from models import AppMode, WorkflowRunTriggeredFrom from models.workflow import Workflow, WorkflowRun from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError from services.workflow_run_service import WorkflowRunListArgs from services.workflow_service import WorkflowService from services.workflow_variable_reference_validator import advisory_variable_reference_warning GRAPH_MODES: Final = (AppMode.WORKFLOW, AppMode.ADVANCED_CHAT) _RUN_LIST_OP: Final = "get.console_app.run" _VERSION_LIST_OP: Final = "get.console_app.version" _RUN_READ_GUARDS: Final = account_app_guards( RBACPermission.APP_CREATE_AND_MANAGEMENT, scope=Scope.APPS_READ, editor=False ) _VERSION_READ_GUARDS: Final = account_app_guards(RBACPermission.APP_VIEW_LAYOUT, scope=Scope.APPS_READ, editor=True) _RELEASE_GUARDS: Final = account_app_guards( RBACPermission.APP_RELEASE_AND_VERSION, scope=Scope.WORKSPACE_WRITE, editor=True ) _ENV_READ_GUARDS: Final = account_app_guards(RBACPermission.APP_VIEW_LAYOUT, scope=Scope.APPS_READ, editor=True) _ENV_WRITE_GUARDS: Final = account_app_guards(RBACPermission.APP_EDIT, scope=Scope.WORKSPACE_WRITE, editor=True) def _request_context(ctx: Context) -> RequestContext: return RequestContext(get_request_id(), get_trace_id(), ctx.account.id, ctx.workspace.id) def _require_uuid(value: str, *, not_found: type[OpenApiError]) -> str: """Parse a path id as a UUID or raise 404; an unparsed id would otherwise reach a UUID-typed database column and fail as a 500.""" try: return str(UUID(value)) except ValueError: raise not_found() def _require_run(ctx: Context, run_id: str) -> WorkflowRun: run = application_services().workflow_runs.get_workflow_run( _request_context(ctx), app_id=ctx.app.id, run_id=_require_uuid(run_id, not_found=RunNotFound) ) if run is None: raise RunNotFound() return run def with_next_cursor(page: RunListResponse, *, op: str, app_id: str, query: RunListQuery) -> RunListResponse: if page.has_more and page.data: next_input = {"app_id": app_id, **query.model_dump(exclude_none=True), "last_id": page.data[-1].id} page.hints = [Hint(summary="Next page", op=op, input=next_input)] return page @openapi_ns.route("/apps//runs") class AppRunListApi(Resource): @endpoint( op=_RUN_LIST_OP, kind=Kind.OBJECT, summary="List runs of a workflow or advanced-chat app, newest first", examples=( Example( title="Latest failed runs from real use", input={"app_id": "", "status": "failed", "triggered_from": "app-run"}, ), Example( title="My latest draft test run", input={"app_id": "", "triggered_from": "debugging", "limit": 1}, ), ), requirements=_RUN_READ_GUARDS, query=RunListQuery, returns=(HTTPStatus.OK, RunListResponse, "Run list"), ) def get(self, ctx: Context, app_id: str, *, query: RunListQuery): require_mode(ctx.app, *GRAPH_MODES) args: WorkflowRunListArgs = {"limit": query.limit} if query.last_id is not None: args["last_id"] = query.last_id if query.status is not None: args["status"] = query.status pagination = application_services().workflow_runs.get_paginate_workflow_runs( _request_context(ctx), app_id=ctx.app.id, args=args, triggered_from=WorkflowRunTriggeredFrom(query.triggered_from or WorkflowRunTriggeredFrom.DEBUGGING), ) page = RunListResponse.model_validate(workflow_run_pagination_response_source(pagination, session=ctx.session)) return with_next_cursor(page, op=_RUN_LIST_OP, app_id=ctx.app.id, query=query) @openapi_ns.route("/apps//runs/") class AppRunDescribeApi(Resource): @endpoint( op="describe.console_app.run", kind=Kind.OBJECT, summary="One run: inputs, outputs, error, tokens and timing", examples=(Example(title="Read one run", input={"app_id": "", "run_id": ""}),), requirements=_RUN_READ_GUARDS, returns=(HTTPStatus.OK, WorkflowRunDetailResponse, "Run detail"), ) def get(self, ctx: Context, app_id: str, run_id: str): require_mode(ctx.app, *GRAPH_MODES) run = _require_run(ctx, run_id) return WorkflowRunDetailResponse.model_validate(workflow_run_response_source(run, session=ctx.session)) @openapi_ns.route("/apps//runs//nodes") class AppRunNodeListApi(Resource): @endpoint( op="get.console_app.run.node", kind=Kind.OBJECT, summary="Each node step of a run, to find the one that failed", examples=(Example(title="Node steps of one run", input={"app_id": "", "run_id": ""}),), requirements=_RUN_READ_GUARDS, returns=(HTTPStatus.OK, WorkflowRunNodeExecutionListResponse, "Node steps"), ) def get(self, ctx: Context, app_id: str, run_id: str): require_mode(ctx.app, *GRAPH_MODES) run = _require_run(ctx, run_id) steps = application_services().workflow_runs.get_workflow_run_node_executions( _request_context(ctx), app_id=ctx.app.id, run_id=run.id ) return WorkflowRunNodeExecutionListResponse.model_validate({"data": steps}) @openapi_ns.route("/apps/:publish") class AppPublishApi(Resource): @endpoint( op="publish.console_app", kind=Kind.OBJECT, summary="Publish the draft as a new version and make the app use it", examples=( Example(title="Publish the draft", input={"app_id": ""}), Example( title="Publish a named version", input={"app_id": "", "marked_name": "v2", "marked_comment": "Adds a retry"}, ), ), requirements=_RELEASE_GUARDS, body=PublishPayload, returns=(HTTPStatus.OK, PublishResponse, "Published"), ) def post(self, ctx: Context, app_id: str, *, body: PublishPayload): require_mode(ctx.app, *GRAPH_MODES) try: with sessionmaker(db.engine).begin() as session: workflow = WorkflowService().publish_app_workflow( session=session, app_model=ctx.app, account=ctx.account, marked_name=body.marked_name, marked_comment=body.marked_comment, ) published = PublishResponse( version_id=workflow.id, created_at=to_timestamp(workflow.created_at), warning=advisory_variable_reference_warning(workflow.graph), ) except ValueError as exc: raise BadRequest(str(exc)) from exc return published @openapi_ns.route("/apps//versions") class AppVersionListApi(Resource): @endpoint( op=_VERSION_LIST_OP, kind=Kind.OBJECT, summary="List published versions, newest first", examples=( Example(title="Latest versions", input={"app_id": ""}), Example(title="Only named versions", input={"app_id": "", "named_only": True}), ), requirements=_VERSION_READ_GUARDS, query=VersionListQuery, returns=(HTTPStatus.OK, VersionListResponse, "Version list"), ) def get(self, ctx: Context, app_id: str, *, query: VersionListQuery): require_mode(ctx.app, *GRAPH_MODES) workflows, has_more = WorkflowService().get_all_published_workflow( session=ctx.session, app_model=ctx.app, page=query.page, limit=query.limit, user_id=None, named_only=query.named_only, include_draft=False, ) page = VersionListResponse( page=query.page, limit=query.limit, has_more=has_more, data=[VersionRow.model_validate(workflow) for workflow in workflows], ) if has_more: next_input = {"app_id": ctx.app.id, **query.model_dump(), "page": query.page + 1} page.hints = [Hint(summary="Next page", op=_VERSION_LIST_OP, input=next_input)] return page @openapi_ns.route("/apps//versions/:restore") class AppVersionRestoreApi(Resource): @endpoint( op="restore.console_app.version", kind=Kind.OBJECT, summary="Copy a published version into the draft; publish again to make it live", examples=( Example(title="Restore an older version", input={"app_id": "", "version_id": ""}), ), requirements=_RELEASE_GUARDS, returns=(HTTPStatus.OK, RestoreResponse, "Restored into the draft"), ) def post(self, ctx: Context, app_id: str, version_id: str): require_mode(ctx.app, *GRAPH_MODES) version_id = _require_uuid(version_id, not_found=VersionNotFound) try: draft = WorkflowService().restore_published_workflow_to_draft( app_model=ctx.app, workflow_id=version_id, account=ctx.account, session=db.session() ) except IsDraftWorkflowError as exc: raise VersionNotRestorable() from exc except WorkflowNotFoundError as exc: raise VersionNotFound() from exc except ValueError as exc: raise BadRequest(str(exc)) from exc return RestoreResponse(result="success", draft_hash=draft.content_hash) def env_variable_rows(variables: Iterable[VariableBase]) -> list[EnvVariableRow]: return [ EnvVariableRow( id=variable.id, name=variable.name, description=variable.description, value_type=environment_variable_value_type(variable), value=encrypter.full_mask_token() if isinstance(variable, SecretVariable) and variable.value else variable.value, ) for variable in variables ] def _stored_secret_ids(draft: Workflow | None) -> set[str]: if draft is None: return set() return {variable.id for variable in draft.environment_variables if isinstance(variable, SecretVariable)} def _patch_env(ctx: Context, *, upserts: list[VariableBase], deletions: list[str]) -> None: try: WorkflowService().patch_draft_workflow_environment_variables( app_model=ctx.app, environment_variables=upserts, deleted_environment_variable_ids=deletions, account=ctx.account, session=db.session(), ) except ValueError as exc: raise BadRequest(str(exc)) from exc @openapi_ns.route("/apps//env") class AppEnvListApi(Resource): @endpoint( op="get.console_app.env", kind=Kind.OBJECT, summary="Environment variables of the draft. Each has an id, which set and delete take, " "and a name, which nodes use. Secrets with a value are masked; empty secrets read as empty", examples=(Example(title="List environment variables", input={"app_id": ""}),), requirements=_ENV_READ_GUARDS, returns=(HTTPStatus.OK, EnvVariableListResponse, "Environment variables"), ) def get(self, ctx: Context, app_id: str): require_mode(ctx.app, *GRAPH_MODES) draft = WorkflowService().get_draft_workflow(app_model=ctx.app, session=ctx.session) if draft is None: raise DraftNotFound() return EnvVariableListResponse(data=env_variable_rows(draft.environment_variables)) @openapi_ns.route("/apps//env/") class AppEnvItemApi(Resource): @endpoint( op="set.console_app.env", kind=Kind.OBJECT, summary="Add or replace one environment variable of the draft, keyed by its id from get console_app env, " "not its name. A new id (e.g. a fresh UUID) adds a variable. Others are left alone", examples=( Example( title="Set an API key as a secret", input={ "app_id": "", "env_id": "", "name": "API_KEY", "value_type": "secret", "value": "sk-...", }, ), ), requirements=_ENV_WRITE_GUARDS, body=EnvVariableSetPayload, returns=(HTTPStatus.OK, SimpleResultResponse, "Variable set"), ) def put(self, ctx: Context, app_id: str, env_id: str, *, body: EnvVariableSetPayload): require_mode(ctx.app, *GRAPH_MODES) if body.value == encrypter.full_mask_token() and body.value_type != EnvVariableValueType.SECRET: raise SecretMaskNotSecret() [mapping] = Workflow.normalize_environment_variable_mappings([{**body.model_dump(mode="json"), "id": env_id}]) if mapping["value"] == HIDDEN_VALUE: draft = WorkflowService().get_draft_workflow(app_model=ctx.app, session=ctx.session) if env_id not in _stored_secret_ids(draft): raise SecretMaskUnknownId() try: variable = variable_factory.build_environment_variable_from_mapping(mapping) except VariableError as exc: raise BadRequest(str(exc)) from exc _patch_env(ctx, upserts=[variable], deletions=[]) return SimpleResultResponse(result="success") @endpoint( op="delete.console_app.env", kind=Kind.OBJECT, summary="Remove one environment variable from the draft, keyed by its id from get console_app env, " "not its name. Others are left alone", examples=(Example(title="Remove a variable", input={"app_id": "", "env_id": ""}),), requirements=_ENV_WRITE_GUARDS, returns=(HTTPStatus.OK, SimpleResultResponse, "Variable removed"), ) def delete(self, ctx: Context, app_id: str, env_id: str): require_mode(ctx.app, *GRAPH_MODES) _patch_env(ctx, upserts=[], deletions=[env_id]) return SimpleResultResponse(result="success")