api/libs/external_api.py
Rama main · réplica de lectura del repositorio original.
import json
import pkgutil
import re
from collections.abc import Callable, Iterable, Mapping
from dataclasses import dataclass
from functools import cache
from typing import Any, ClassVar, Protocol, TypedDict, override
from flask import Blueprint, Flask, current_app, got_request_exception, request
from flask.typing import ResponseReturnValue
from flask_restx import Api
from pydantic import BaseModel
from pydantic.json_schema import SkipJsonSchema
from werkzeug.datastructures import Headers
from werkzeug.exceptions import HTTPException
from werkzeug.http import HTTP_STATUS_CODES
from configs import dify_config
from core.errors.error import AppInvokeQuotaExceededError
from core.plugin.impl.exc import PluginRuntimeError
from extensions.ext_logging import get_request_id
from libs.exception import BaseHTTPException, RelayedHTTPError
from libs.flask_restx_compat import install_swagger_compatibility
from libs.token import build_force_logout_cookie_headers
def http_status_message(code):
return HTTP_STATUS_CODES.get(code, "")
class ErrorResponse(BaseModel):
"""Documented non-2xx body emitted by ``register_external_error_handlers``.
``params`` and ``details`` appear only on the errors that set them. The schema
stays open because ``HTTPException.data`` may carry extra keys.
"""
code: str
message: str
status: int
params: str | SkipJsonSchema[None] = None
details: dict[str, Any] | SkipJsonSchema[None] = None
_ERROR_SCHEMA_REF_TEMPLATE = "#/components/schemas/{model}"
ERROR_CODES_VENDOR_KEY = "dify-error-codes"
_ERROR_CODES_EXTENSION = f"x-{ERROR_CODES_VENDOR_KEY}"
# Marks an operation that relays downstream errors: any code may answer it with the error body.
RELAYS_ERRORS_EXTENSION = "x-dify-relays-errors"
# ``BaseHTTPException`` ships this placeholder; a documented business error must replace it.
_UNDECLARED_ERROR_CODE = BaseHTTPException.error_code
class ErrorSpec(TypedDict):
"""How one documented exception reaches the wire, before a surface picks its ``code``."""
status: int
# The class's own ``error_code``, if it declares one.
error_code: str | None
# The ``code`` the surface's handlers emit when no body formatter rewrites it.
default_code: str
# What the non-HTTP handlers in register_external_error_handlers emit, as (status, code).
_SHARED_HANDLED_ERRORS: Mapping[type[Exception], tuple[int, str]] = {
ValueError: (400, "invalid_param"),
AppInvokeQuotaExceededError: (429, "too_many_requests"),
PluginRuntimeError: (502, "plugin_runtime_error"),
}
def _class_attribute(cls: type, name: str) -> object:
"""``name`` as the class or its nearest base defines it, without instance or ``__getattr__`` lookups."""
for klass in cls.__mro__:
if name in klass.__dict__:
return klass.__dict__[name]
return None
def _snake_case(name: str) -> str:
return re.sub(r"(?<!^)(?=[A-Z])", "_", name).lower()
@cache
def _framework_error_codes() -> Mapping[int, frozenset[str]]:
"""The codes Werkzeug's own HTTP exceptions produce, by status.
The HTTP handler names the code after the raised class, so subclasses such as
``BadRequestKeyError`` (a missing form or file field) answer with their own code.
Werkzeug defines all of them at import, so the walk happens once.
"""
by_status: dict[int, set[str]] = {}
pending: list[type[HTTPException]] = [HTTPException]
while pending:
exception = pending.pop()
pending.extend(exception.__subclasses__())
if exception.__module__.startswith("werkzeug.") and isinstance(exception.code, int) and exception.code >= 400:
by_status.setdefault(exception.code, set()).add(_snake_case(exception.__name__))
return {status: frozenset(codes) for status, codes in by_status.items()}
def _error_spec(
exception: type[Exception],
handled: Mapping[type[Exception], tuple[int, str]],
dispatch_order: Iterable[type[BaseException]],
) -> ErrorSpec:
declared = _class_attribute(exception, "error_code")
error_code = str(declared) if isinstance(declared, str) else None
# Flask-RESTX dispatches to the first registered handler whose type matches, in
# registration order (not the most specific one), so resolve the same way.
for handler_type in dispatch_order:
if not issubclass(exception, handler_type):
continue
if handler_type in handled:
status, default_code = handled[handler_type]
return {"status": status, "error_code": error_code, "default_code": default_code}
if handler_type is HTTPException and issubclass(exception, HTTPException):
if exception.code is None:
raise ValueError(f"{exception.__name__} declares no HTTP status")
if issubclass(exception, BaseHTTPException):
default_code = exception.error_code
else:
default_code = _snake_case(exception.__name__)
return {"status": exception.code, "error_code": error_code, "default_code": default_code}
break
raise ValueError(f"{exception.__name__} has no dedicated error handler; its status and code are not stable")
def keeps_placeholder_error_code(exception: type[Exception]) -> bool:
"""Whether ``exception`` would answer with ``BaseHTTPException``'s placeholder ``code``.
Such an error cannot be documented: declare an ``error_code`` on it. Relayed errors
take their code at runtime and are exempt.
"""
return (
issubclass(exception, BaseHTTPException)
and not issubclass(exception, RelayedHTTPError)
and _class_attribute(exception, "error_code") == _UNDECLARED_ERROR_CODE
)
def is_business_error(exception: type[Exception]) -> bool:
"""Whether ``exception`` names an operation-specific outcome worth listing.
Werkzeug exceptions and anything dispatched through a dedicated (non-HTTP) handler
are generic instead: clients handle them centrally, and every operation's open
``default`` error response covers them.
"""
return issubclass(exception, HTTPException) and not exception.__module__.startswith("werkzeug.")
_HTTP_METHOD_DOC_KEYS = ("get", "put", "post", "delete", "patch", "head", "options", "trace")
def _with_documented_errors(doc: Mapping[str, Any], names: Iterable[str]) -> dict[str, Any]:
vendor = dict(doc.get("vendor", {}))
# A dict, not a list: Flask-RESTX deep-merges dicts across decorators and class/method
# docs but replaces lists, so every contributor's errors survive the merge.
vendor[ERROR_CODES_VENDOR_KEY] = {**vendor.get(ERROR_CODES_VENDOR_KEY, {}), **dict.fromkeys(names, True)}
return {**doc, "vendor": vendor}
def _documented_names(exceptions: Iterable[type[Exception]]) -> list[str]:
exceptions = list(exceptions)
generic = [exception.__name__ for exception in exceptions if not is_business_error(exception)]
if generic:
raise ValueError(f"generic errors are implied, do not list them: {', '.join(generic)}")
undeclared = [exception.__name__ for exception in exceptions if keeps_placeholder_error_code(exception)]
if undeclared:
raise ValueError(f"business errors must declare an error_code: {', '.join(undeclared)}")
return [f"{exception.__module__}:{exception.__qualname__}" for exception in exceptions]
def document_errors[T](target: T, *exceptions: type[Exception]) -> T:
"""Record the business errors a view or Resource class can raise, for ``error_responses`` docs.
A view decorator calls this on the wrapper it returns, so every operation it guards
documents the errors it raises. On a Resource class (for ``method_decorators``) the
errors apply to each of its operations. Only business errors are accepted; generic
ones are implied (see ``is_business_error``). Builds new doc dicts because
``functools.wraps`` shares the wrapped view's ``__apidoc__``.
"""
names = _documented_names(exceptions)
existing = _class_attribute(target, "__apidoc__") if isinstance(target, type) else vars(target).get("__apidoc__")
if existing is False:
# ``@ns.hide`` hides the view or resource from the document; there is nothing to document.
return target
doc: Mapping[str, Any] = existing if isinstance(existing, Mapping) else {}
if isinstance(target, type):
# Class-level docs only reach operations through their per-method entries. A method
# hidden with ``@ns.doc(<method>=False)`` stays hidden.
doc = {
**doc,
**{
method: entry
if entry is False
else _with_documented_errors(entry if isinstance(entry, Mapping) else {}, names)
for method in _HTTP_METHOD_DOC_KEYS
for entry in (doc.get(method),)
},
}
else:
doc = _with_documented_errors(doc, names)
target.__apidoc__ = doc # type: ignore[attr-defined]
return target
class ViewDecorator(Protocol):
"""A decorator that returns the view it is given, keeping the view's type."""
def __call__[F: Callable[..., object]](self, view: F, /) -> F: ...
def error_responses(*exceptions: type[Exception]) -> ViewDecorator:
"""Document the business errors an operation's handler raises.
List what the handler and its callees raise; decorators document their own errors
with ``document_errors``. Generic errors are never listed: every operation documents
an open ``default`` error for them. Each surface resolves status and ``code`` from its
own handlers when it builds the document, and documents each code once as a shared
``{Code}Error`` component, so a business error must declare an ``error_code``.
Nothing infers this list from the code: list what the operation's own call chain can
answer with, not everything a shared translator knows how to map. The controller unit
tests check every served error response against the document.
"""
_documented_names(exceptions)
def decorator[F: Callable[..., object]](view: F, /) -> F:
return document_errors(view, *exceptions)
return decorator
def _pascal_case(value: str) -> str:
return "".join(word[:1].upper() + word[1:] for word in re.split(r"[^0-9A-Za-z]+", value) if word)
def error_component_name(code: str) -> str:
"""The component that documents ``code``: ``draft_workflow_not_sync`` becomes ``DraftWorkflowNotSyncError``."""
name = _pascal_case(code)
return name if name.endswith("Error") else f"{name}Error"
def _details_model(exception: type[Exception]) -> type[BaseModel] | None:
"""The Pydantic model an exception declares for its ``details``, if any."""
model = _class_attribute(exception, "details_model")
return model if isinstance(model, type) and issubclass(model, BaseModel) else None
@dataclass(frozen=True)
class _ErrorComponent:
code: str
status: int
exception: type[Exception]
details_model: type[BaseModel] | None
where: str
def schema(self, name: str, error_body_ref: Mapping[str, object]) -> dict[str, object]:
properties: dict[str, object] = {
"code": {"const": self.code, "type": "string"},
"status": {"const": self.status, "type": "integer"},
}
if self.details_model is not None:
properties["details"] = {"$ref": _ERROR_SCHEMA_REF_TEMPLATE.format(model=self.details_model.__name__)}
schema: dict[str, object] = {
"allOf": [
dict(error_body_ref),
{"properties": properties, "required": ["code", "status"], "type": "object"},
],
"title": name,
}
description = _class_attribute(self.exception, "description")
if isinstance(description, str) and description:
schema["description"] = description
return schema
def _explicit_json_schema(response: Mapping[str, Any]) -> object | None:
"""The JSON body schema a response already declares, if any; other media types are not errors."""
content = response.get("content")
media = content.get("application/json") if isinstance(content, dict) else None
return media.get("schema") if isinstance(media, dict) else None
def _comparable_schema(schema: object) -> object:
"""``schema`` with order-insensitive parts (``required``, ``enum``) normalized for equality checks."""
if isinstance(schema, dict):
return {
key: sorted(value, key=lambda item: json.dumps(item, sort_keys=True))
if key in {"required", "enum"} and isinstance(value, list)
else _comparable_schema(value)
for key, value in schema.items()
}
if isinstance(schema, list):
return [_comparable_schema(item) for item in schema]
return schema
def _add_model_schemas(schemas: dict[str, object], model: type[BaseModel]) -> None:
"""Add ``model`` and its nested definitions; a name already documenting a different schema is an error."""
schema = model.model_json_schema(ref_template=_ERROR_SCHEMA_REF_TEMPLATE, mode="serialization")
definitions = schema.pop("$defs", {})
for name, definition in {**definitions, model.__name__: schema}.items():
existing = schemas.setdefault(name, definition)
if _comparable_schema(existing) != _comparable_schema(definition):
raise ValueError(
f"schema {name} (from {model.__name__}) collides with a different component of the same name"
)
def attach_error_response_schema(
document: dict[str, object],
model: type[BaseModel],
resolve_error: Callable[[type[Exception]], tuple[int, str]],
is_generic: Callable[[type[Exception]], bool] = lambda exception: False,
) -> dict[str, object]:
"""Give every 4xx/5xx response the surface's error body and document business errors by code.
Every operation gets a ``default`` error response with the open error body: generic
errors (authentication, validation, rate limits, unexpected failures) can answer any
operation and clients handle them centrally, so no operation lists them. ``is_generic``
names the surface's own generic errors (see ``ExternalApi.document_generic_errors``);
an operation that documents one gets nothing for it beyond the ``default`` error.
Each business error an operation documents with ``error_responses`` is one shared
``{Code}Error`` component: the error body with ``code`` and ``status`` fixed, plus the
class's ``details_model`` when it declares one. The response of its status lists it in
an ``anyOf`` that always ends with the open error body, since a generic error can answer
that status too and ``default`` only covers statuses the operation leaves undeclared; a
status that also declares its own body schema keeps that schema in front of them. A code
maps to exactly one status in a document, and two classes may share a code only when
they agree on it. A documented
``RelayedHTTPError`` adds no component, since its status and code are only known at
runtime; it marks the operation with ``x-dify-relays-errors`` instead.
"""
paths = document.get("paths")
if not isinstance(paths, dict):
return document
ref: dict[str, object] = {"$ref": _ERROR_SCHEMA_REF_TEMPLATE.format(model=model.__name__)}
error_components: dict[str, _ErrorComponent] = {}
used = False
for path, path_item in paths.items():
if not isinstance(path_item, dict):
continue
for method, operation in path_item.items():
if not isinstance(operation, dict):
continue
exception_names: Mapping[str, object] = operation.pop(_ERROR_CODES_EXTENSION, {})
responses = operation.setdefault("responses", {})
if not isinstance(responses, dict):
continue
where = f"{method.upper()} {path}"
components_by_status: dict[str, set[str]] = {}
for exception_name in exception_names:
exception = pkgutil.resolve_name(exception_name)
if is_generic(exception):
# Covered by the operation's open ``default`` error.
continue
if issubclass(exception, RelayedHTTPError):
# Its status and code come from a downstream service at runtime; the open
# ``default`` error already documents it. The marker lets checks that compare
# served bodies with the document admit any code here.
operation[RELAYS_ERRORS_EXTENSION] = True
continue
status, code = resolve_error(exception)
name = error_component_name(code)
component = _ErrorComponent(code, status, exception, _details_model(exception), where)
known = error_components.setdefault(name, component)
if known.code != code:
raise ValueError(
f"error codes {known.code!r} ({known.where}) and {code!r} ({where}) both document as {name}"
)
if known.status != status:
raise ValueError(
f"error code {code!r} answers {status} on {where} but {known.status} on {known.where}; "
"a code maps to one status"
)
if known.details_model is not component.details_model:
raise ValueError(
f"error code {code!r} declares different details models on {known.where} and {where}"
)
components_by_status.setdefault(str(status), set()).add(name)
for status, names in components_by_status.items():
response = responses.setdefault(status, {"description": http_status_message(int(status))})
if not isinstance(response, dict):
continue
options: list[object] = [
{"$ref": _ERROR_SCHEMA_REF_TEMPLATE.format(model=name)} for name in sorted(names)
]
explicit = _explicit_json_schema(response)
if explicit is not None:
# The declared body and the documented errors can all answer this status.
options.insert(0, explicit)
# Generic errors (validation, auth) answer this status as well, and ``default``
# does not apply to a status the operation declares, so the open body stays.
options.append(dict(ref))
content = response.get("content")
if not isinstance(content, dict):
content = response["content"] = {}
media = content.get("application/json")
# Other media types the status declares (a binary or text body) stay as they are.
json_media = media if isinstance(media, dict) else {}
content["application/json"] = {**json_media, "schema": {"anyOf": options}}
used = True
responses.setdefault("default", {"description": "Error"})
for status, response in responses.items():
if status in components_by_status or not (status == "default" or str(status).startswith(("4", "5"))):
continue
if not isinstance(response, dict):
continue
content = response.get("content")
if not isinstance(content, dict) or not content:
response["content"] = {"application/json": {"schema": dict(ref)}}
used = True
continue
media = content.get("application/json")
if not isinstance(media, dict):
continue
declared = media.get("schema")
if declared is None:
media["schema"] = dict(ref)
used = True
elif declared != ref:
# A declared body never replaces the error body: the surface's handlers can
# still answer this status with it (validation, auth, unexpected errors).
media["schema"] = {"anyOf": [declared, dict(ref)]}
used = True
if not used:
return document
components = document.setdefault("components", {})
if not isinstance(components, dict):
return document
schemas = components.setdefault("schemas", {})
if not isinstance(schemas, dict):
return document
_add_model_schemas(schemas, model)
for name, component in error_components.items():
if component.details_model is not None:
_add_model_schemas(schemas, component.details_model)
if name in schemas:
raise ValueError(f"error component {name} (code {component.code!r}) collides with an existing component")
schemas[name] = component.schema(name, ref)
return document
class ErrorBodyFormatter(Protocol):
"""Last-touch hook over an error body before it goes on the wire."""
def finalize(self, e: Exception, data: dict[str, Any], status_code: int) -> dict[str, Any]: ...
def register_external_error_handlers(api: Api, body_formatter: ErrorBodyFormatter | None = None):
def _finalize(e: Exception, data: dict[str, Any], status_code: int) -> dict[str, Any]:
if body_formatter is None:
return data
return body_formatter.finalize(e, data, status_code)
def handle_http_exception(e: HTTPException):
got_request_exception.send(current_app, exception=e)
# If Werkzeug already prepared a Response, just use it. This bypasses
# body_formatter entirely — surfaces with a formatter must not raise
# exceptions carrying a pre-built response.
if e.response is not None:
return e.response
status_code = getattr(e, "code", 500) or 500
# Build a safe, dict-like payload
default_data = {
"code": _snake_case(type(e).__name__),
"message": getattr(e, "description", http_status_message(status_code)),
"status": status_code,
}
if default_data["message"] == "Failed to decode JSON object: Expecting value: line 1 column 1 (char 0)":
default_data["message"] = "Invalid JSON payload received or JSON payload is empty."
# Preserve protocol headers while rendering our own JSON body.
headers = Headers(e.get_headers())
headers.remove("Content-Type")
headers.remove("Content-Length")
exc_headers = getattr(e, "headers", None)
if exc_headers:
headers.update(Headers(exc_headers))
# Payload per status
if status_code == 406 and api.default_mediatype is None:
data = {"code": "not_acceptable", "message": default_data["message"], "status": status_code}
return _finalize(e, data, status_code), status_code, headers
elif status_code == 400:
msg = default_data["message"]
if isinstance(msg, Mapping) and msg:
# Convert param errors like {"field": "reason"} into a friendly shape
param_key, param_value = next(iter(msg.items()))
data = {
"code": "invalid_param",
"message": str(param_value),
"params": param_key,
"status": status_code,
}
else:
data = {**default_data}
data.setdefault("code", "unknown")
return _finalize(e, data, status_code), status_code, headers
else:
data = {**default_data}
data.setdefault("code", "unknown")
# If you need WWW-Authenticate for 401, add it to headers
if status_code == 401:
headers.setdefault("WWW-Authenticate", 'Bearer realm="api"')
# Check if this is a forced logout error - clear cookies
error_code = getattr(e, "error_code", None)
if error_code == "unauthorized_and_force_logout":
# Add Set-Cookie headers to clear auth cookies
headers.setlist("Set-Cookie", build_force_logout_cookie_headers())
return _finalize(e, data, status_code), status_code, headers
def handle_value_error(e: ValueError):
got_request_exception.send(current_app, exception=e)
current_app.logger.exception("value_error in request handler")
status_code = 400
data = {"code": "invalid_param", "message": str(e), "status": status_code}
return _finalize(e, data, status_code), status_code
def handle_quota_exceeded(e: AppInvokeQuotaExceededError):
got_request_exception.send(current_app, exception=e)
status_code = 429
data = {"code": "too_many_requests", "message": str(e), "status": status_code}
return _finalize(e, data, status_code), status_code
def handle_plugin_runtime_error(e: PluginRuntimeError):
got_request_exception.send(current_app, exception=e)
status_code = 502
details = {"request_id": get_request_id()}
if e.lambda_request_id:
details["lambda_request_id"] = e.lambda_request_id
data = {
"code": "plugin_runtime_error",
"message": e.description,
"details": details,
"status": status_code,
}
return _finalize(e, data, status_code), status_code
def handle_general_exception(e: Exception):
got_request_exception.send(current_app, exception=e)
status_code = 500
data: dict[str, Any] = getattr(e, "data", {"message": http_status_message(status_code)})
# 🔒 Normalize non-mapping data (e.g., if someone set e.data = Response)
if not isinstance(data, dict):
data = {"message": str(e)}
data.setdefault("code", "unknown")
data.setdefault("status", status_code)
# Note: Exception logging is handled by Flask/Flask-RESTX framework automatically
# Explicit log_exception call removed to avoid duplicate log entries
return _finalize(e, data, status_code), status_code
api.errorhandler(HTTPException)(handle_http_exception) # pyrefly: ignore[not-callable] - Flask-RESTX typing
api.errorhandler(ValueError)(handle_value_error) # pyrefly: ignore[not-callable] - Flask-RESTX typing
api.errorhandler(AppInvokeQuotaExceededError)(handle_quota_exceeded) # pyrefly: ignore[not-callable] - Flask-RESTX typing
api.errorhandler(PluginRuntimeError)(handle_plugin_runtime_error) # pyrefly: ignore[not-callable] - Flask-RESTX typing
api.errorhandler(Exception)(handle_general_exception) # pyrefly: ignore[not-callable] - Flask-RESTX typing
class ExternalApi(Api):
error_response_model: ClassVar[type[BaseModel]] = ErrorResponse
_authorizations = {
"Bearer": {
"bearerFormat": "API_KEY",
"description": "Use the Service API key as a Bearer token in the Authorization header.",
"scheme": "bearer",
"type": "http",
}
}
def __init__(
self,
app: Blueprint | Flask,
*args,
error_body_formatter: ErrorBodyFormatter | None = None,
register_default_root: bool = True,
**kwargs,
):
self._error_body_formatter = error_body_formatter
self._handled_errors: dict[type[Exception], tuple[int, str]] = dict(_SHARED_HANDLED_ERRORS)
self._generic_errors: list[type[Exception]] = []
self._implicit_codes: dict[int, frozenset[str]] = {}
self._register_default_root = register_default_root
install_swagger_compatibility()
kwargs.setdefault("authorizations", self._authorizations)
kwargs.setdefault("security", "Bearer")
kwargs["add_specs"] = dify_config.SWAGGER_UI_ENABLED
kwargs["doc"] = dify_config.SWAGGER_UI_PATH if dify_config.SWAGGER_UI_ENABLED else False
if error_body_formatter is not None:
kwargs.setdefault("catch_all_404s", True)
# the override below patches a private flask-restx method; fail at
# startup (not at the first 404) if an upgrade removes it
if not callable(getattr(Api, "_should_use_fr_error_handler", None)):
raise RuntimeError("flask-restx no longer exposes _should_use_fr_error_handler; update ExternalApi")
# manual separate call on construction and init_app to ensure configs in kwargs effective
super().__init__(*args, app=None, **kwargs)
self.init_app(app, **kwargs)
register_external_error_handlers(self, body_formatter=error_body_formatter)
def finish_document(self, document: dict[str, object]) -> dict[str, object]:
"""A surface overrides this to rewrite its finished OpenAPI document (see
``DocumentFinisher`` in ``flask_restx_compat``); overrides must call ``super()``.
The default documents ``error_response_model`` on error responses."""
return attach_error_response_schema(
document, self.error_response_model, self.resolve_error, self.is_generic_error
)
def document_error_handler(self, exception: type[Exception], *, status: int, code: str) -> None:
"""Record what a surface-specific handler for ``exception`` emits, for ``error_responses``."""
self._handled_errors[exception] = (status, code)
self._implicit_codes.clear()
def document_generic_errors(self, *exceptions: type[Exception]) -> None:
"""Declare business-shaped errors this surface's clients handle centrally.
A generic error is one any operation may answer with and no caller branches on
per operation: a missing setup step, an uninitialized account, a revoked
license, a malformed argument. It is covered by every operation's open
``default`` error and admitted by ``implicit_error_codes``; an operation that
documents it gets no per-operation entry for it. Register them once, where the
surface is built, so the decision is visible in one place.
"""
for exception in exceptions:
if not is_business_error(exception):
raise ValueError(f"{exception.__name__} is generic already; only business-shaped errors are declared")
if keeps_placeholder_error_code(exception):
raise ValueError(f"{exception.__name__} must declare an error_code to be documented as generic")
self.resolve_error(exception)
if exception not in self._generic_errors:
self._generic_errors.append(exception)
self._implicit_codes.clear()
def is_generic_error(self, exception: type[Exception]) -> bool:
"""Whether this surface documents ``exception`` as generic (see ``document_generic_errors``)."""
return any(issubclass(exception, generic) for generic in self._generic_errors)
def resolve_error(self, exception: type[Exception]) -> tuple[int, str]:
"""The (status, ``code``) this surface emits for ``exception``.
Raises ``ValueError`` when no handler gives it a stable status and code.
"""
spec = _error_spec(exception, self._handled_errors, self._error_dispatch_order())
return spec["status"], self.error_code_for(spec)
def _error_dispatch_order(self) -> list[type[BaseException]]:
"""Handler types in the order Flask-RESTX tries them: the API's own, then new namespace ones."""
order: list[type[BaseException]] = list(self.error_handlers)
for namespace in self.namespaces:
order.extend(exception for exception in namespace.error_handlers if exception not in order)
return order
def error_code_for(self, spec: ErrorSpec) -> str:
"""The ``code`` this surface emits for a documented exception."""
return spec["default_code"]
def implicit_error_codes(self, status: int) -> set[str]:
"""Generic codes any operation may emit at ``status`` without listing them.
The document covers them with each operation's open ``default`` error; checks that
compare served error bodies with the document admit them beside the documented
business codes. The result is cached per status until the surface declares another
handler or generic error, since checks ask for it on every error response.
"""
cached = self._implicit_codes.get(status)
if cached is not None:
return set(cached)
codes = set(_framework_error_codes().get(status, ()))
codes.update(code for handled_status, code in self._handled_errors.values() if handled_status == status)
for exception in self._generic_errors:
generic_status, code = self.resolve_error(exception)
if generic_status == status:
codes.add(code)
if status == 500:
# Unhandled exceptions reach the catch-all handler.
codes.add("unknown")
self._implicit_codes[status] = frozenset(codes)
return codes
@override
def _register_doc(self, app_or_blueprint: Blueprint | Flask) -> None:
if self._add_specs and self._doc:
app_or_blueprint.add_url_rule(self._doc, "doc", self.render_doc)
# Api.base_path resolves the ``root`` endpoint. A caller that disables
# Flask-RESTX's 404 root must register its own resource with that endpoint.
if self._register_default_root:
def render_default_root() -> ResponseReturnValue:
self.render_root()
return "", 404
app_or_blueprint.add_url_rule(self.prefix or "/", "root", render_default_root)
@override
def _should_use_fr_error_handler(self):
# catch_all_404s makes flask-restx claim NotFound for ANY app path
# (it wraps the app-level handle_exception), so scope the claim to
# this blueprint's url prefix; other surfaces keep their own 404s.
if self._error_body_formatter is not None and not self._request_under_own_prefix():
return False
return super()._should_use_fr_error_handler()
def _request_under_own_prefix(self) -> bool:
prefix = self.blueprint.url_prefix if self.blueprint is not None else None
if not prefix:
return True
return request.path == prefix or request.path.startswith(prefix.rstrip("/") + "/")