import json
from typing import Annotated
from fastapi import APIRouter
from fastapi import Body
from fastapi import Path
from fastapi import Query
from fastapi.responses import JSONResponse
from ...backends import json_backend
from ...config import EwoksSettingsType
from .. import status
from ..common import discovery
from ..common import models as common_models
from . import backend
from . import descriptions
from . import models
v1_0_0_router = APIRouter()
[docs]
@v1_0_0_router.get(
"/workflow/{identifier}",
summary="Get ewoks workflow",
response_model=models.EwoksWorkflow,
response_model_exclude_none=True,
response_description="Ewoks workflow",
status_code=200,
responses={
status.HTTP_403_FORBIDDEN: {
"description": "No permission to read workflow",
"model": common_models.ResourceIdentifierError,
},
status.HTTP_404_NOT_FOUND: {
"description": "Workflow not found",
"model": common_models.ResourceIdentifierError,
},
},
)
def get_workflow(
identifier: Annotated[
str,
Path(
title="Workflow identifier",
description="Unique identifier in the workflow database",
),
],
settings: EwoksSettingsType,
) -> json_backend.ResourceContentType:
try:
return backend.load_workflow(
settings, settings.resource_directory / "workflows", identifier
)
except PermissionError:
return JSONResponse(
{
"message": f"No permission to read workflow '{identifier}'.",
"type": "workflow",
"identifier": identifier,
},
status_code=status.HTTP_403_FORBIDDEN,
)
except FileNotFoundError:
return JSONResponse(
{
"message": f"Workflow '{identifier}' is not found.",
"type": "workflow",
"identifier": identifier,
},
status_code=status.HTTP_404_NOT_FOUND,
)
[docs]
@v1_0_0_router.get(
"/workflows",
summary="Get all ewoks workflow identifiers",
response_model=models.EwoksWorkflowIdentifiers,
response_description="Ewoks workflow identifiers",
status_code=200,
)
def get_workflow_identifiers(
settings: EwoksSettingsType, kw: Annotated[list[str] | None, Query()] = None
) -> dict[str, list[str]]:
keywords = _compile_keywords(kw)
root = settings.resource_directory / "workflows"
if keywords:
identifiers = [
desc["id"]
for desc in descriptions.workflow_descriptions(
settings, root, keywords=keywords
)
]
else:
identifiers = backend.workflow_identifiers(settings, root)
return {"identifiers": identifiers}
def _compile_keywords(kw: list[str] | None) -> dict | None:
if not kw:
return
keywords = dict()
for s in kw:
name, _, value = s.partition(":")
try:
value = json.loads(value)
except json.decoder.JSONDecodeError:
pass
keywords[name] = value
return keywords
[docs]
@v1_0_0_router.get(
"/workflows/descriptions",
summary="Get all ewoks workflow descriptions",
response_model=models.EwoksWorkflowDescriptions,
response_model_exclude_none=True,
response_description="Ewoks workflow descriptions",
status_code=200,
)
def get_workflows(
settings: EwoksSettingsType, kw: Annotated[list[str] | None, Query()] = None
) -> dict[str, list[dict]]:
keywords = _compile_keywords(kw)
return {
"items": list(
descriptions.workflow_descriptions(
settings, settings.resource_directory / "workflows", keywords=keywords
)
)
}
[docs]
@v1_0_0_router.put(
"/workflow/{identifier}",
summary="Update ewoks workflow",
response_model=models.EwoksWorkflow,
response_model_exclude_none=True,
response_description="Ewoks workflow",
status_code=200,
responses={
status.HTTP_400_BAD_REQUEST: {
"description": "Wrong workflow identifier",
"model": common_models.ResourceIdentifierError,
},
status.HTTP_404_NOT_FOUND: {
"description": "Workflow not found",
"model": common_models.ResourceIdentifierError,
},
status.HTTP_403_FORBIDDEN: {
"description": "No permission to edit workflow",
"model": common_models.ResourceIdentifierError,
},
},
)
def update_workflow(
identifier: Annotated[
str,
Path(
title="Workflow identifier",
description="Unique identifier in the workflow database",
),
],
workflow: Annotated[models.EwoksWorkflow, Body(title="Ewoks workflow")],
settings: EwoksSettingsType,
) -> models.EwoksWorkflow:
if workflow.graph:
ridentifier = workflow.graph.get("id", identifier)
else:
ridentifier = None
if identifier != ridentifier:
return JSONResponse(
{
"message": f"Resource identifier '{identifier}' is not equal to '{ridentifier}'.",
"type": "workflow",
"identifier": identifier,
},
status_code=status.HTTP_400_BAD_REQUEST,
)
exists = backend.workflow_exists(
settings, settings.resource_directory / "workflows", identifier
)
if not exists:
return JSONResponse(
{
"message": f"Workflow '{identifier}' is not found.",
"type": "workflow",
"identifier": identifier,
},
status_code=status.HTTP_404_NOT_FOUND,
)
try:
backend.save_workflow(
settings,
settings.resource_directory / "workflows",
identifier,
workflow.model_dump(exclude_none=True),
)
except PermissionError:
return JSONResponse(
{
"message": f"No permission to edit workflow '{identifier}'.",
"type": "workflow",
"identifier": identifier,
},
status_code=status.HTTP_403_FORBIDDEN,
)
return workflow
[docs]
@v1_0_0_router.post(
"/workflows",
summary="Create ewoks workflow",
response_model=models.EwoksWorkflow,
response_model_exclude_none=True,
response_description="Ewoks workflow",
status_code=200,
responses={
status.HTTP_400_BAD_REQUEST: {
"description": "Workflow identifier missing",
"model": common_models.ResourceIdentifierError,
},
status.HTTP_422_UNPROCESSABLE_CONTENT: {
"description": "Workflow identifier cannot be empty",
"model": common_models.ResourceIdentifierError,
},
status.HTTP_409_CONFLICT: {
"description": "Workflow already exists",
"model": common_models.ResourceIdentifierError,
},
status.HTTP_403_FORBIDDEN: {
"description": "No permission to create workflow",
"model": common_models.ResourceIdentifierError,
},
},
)
def create_workflow(
workflow: Annotated[models.EwoksWorkflow, Body(title="Ewoks workflow")],
settings: EwoksSettingsType,
) -> models.EwoksWorkflow:
if workflow.graph:
ridentifier = workflow.graph.get("id")
else:
ridentifier = None
if ridentifier is None:
return JSONResponse(
{
"message": "Workflow identifier missing",
"type": "workflow",
},
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
)
if ridentifier == "":
return JSONResponse(
{
"message": "Workflow identifier cannot be empty",
"type": "workflow",
},
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
)
try:
exists = backend.workflow_exists(
settings, settings.resource_directory / "workflows", ridentifier
)
except ValueError:
return JSONResponse(
{
"message": f"Workflow identifier '{ridentifier}' is not valid",
"type": "workflow",
"identifier": ridentifier,
},
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
)
if exists:
return JSONResponse(
{
"message": f"Workflow '{ridentifier}' already exists.",
"type": "workflow",
"identifier": ridentifier,
},
status_code=status.HTTP_409_CONFLICT,
)
try:
backend.save_workflow(
settings,
settings.resource_directory / "workflows",
ridentifier,
workflow.model_dump(exclude_none=True),
)
except PermissionError:
return JSONResponse(
{
"message": f"No permission to create workflow '{ridentifier}'.",
"type": "workflow",
"identifier": ridentifier,
},
status_code=status.HTTP_403_FORBIDDEN,
)
return workflow
[docs]
@v1_0_0_router.delete(
"/workflow/{identifier}",
summary="Delete ewoks workflow",
response_model=common_models.ResourceInfo,
response_description="Deleted ewoks workflow",
status_code=200,
responses={
status.HTTP_403_FORBIDDEN: {
"description": "No permission to delete workflow, or the workflow "
"is not managed by ewoksserver and cannot be deleted",
"model": common_models.ResourceIdentifierError,
},
status.HTTP_404_NOT_FOUND: {
"description": "Workflow not found",
"model": common_models.ResourceIdentifierError,
},
},
)
def delete_workflow(
identifier: Annotated[
str,
Path(
title="Workflow identifier",
description="Unique identifier in the workflow database",
),
],
settings: EwoksSettingsType,
) -> dict[str, str]:
if backend.is_external_workflow(settings, identifier):
return JSONResponse(
{
"message": f"Workflow '{identifier}' is not managed by ewoksserver "
"and cannot be deleted.",
"type": "workflow",
"identifier": identifier,
},
status_code=status.HTTP_403_FORBIDDEN,
)
try:
backend.delete_workflow(settings.resource_directory / "workflows", identifier)
except PermissionError:
return JSONResponse(
{
"message": f"No permission to delete workflow '{identifier}'.",
"type": "workflow",
"identifier": identifier,
},
status_code=status.HTTP_403_FORBIDDEN,
)
except FileNotFoundError:
return JSONResponse(
{
"message": f"Workflow '{identifier}' is not found.",
"type": "workflow",
"identifier": identifier,
},
status_code=status.HTTP_404_NOT_FOUND,
)
return {"identifier": identifier, "type": "workflow"}
v2_1_0_router = APIRouter()
v2_1_0_router.include_router(v1_0_0_router)
[docs]
@v2_1_0_router.post(
"/workflows/discover",
summary="Discover ewoks workflow identifiers from a worker environment",
response_model=models.EwoksWorkflowIdentifiers,
response_description="Discovered ewoks workflow identifiers",
status_code=200,
responses={
status.HTTP_404_NOT_FOUND: {
"description": "Module not found",
"model": common_models.ResourceError,
},
},
)
def discover_workflows(
settings: EwoksSettingsType,
options: Annotated[
models.EwoksWorkflowDiscovery, Body(title="Ewoks workflow discovery options")
] = None,
) -> dict[str, list[str]]:
if options:
discover_options = options.model_dump()
else:
discover_options = dict()
try:
identifiers, identifier_to_queue = discovery.discover_workflows(
settings, **discover_options
)
except ModuleNotFoundError as e:
return JSONResponse(
{
"message": str(e),
"type": "workflow",
},
status_code=status.HTTP_404_NOT_FOUND,
)
backend.register_external_workflows(
settings,
settings.resource_directory / "workflows",
identifier_to_queue,
worker_options=discover_options.get("worker_options"),
)
return {"identifiers": identifiers}
