diff --git a/http_api_cerberus/README.rst b/http_api_cerberus/README.rst new file mode 100644 index 00000000..b6790cf4 --- /dev/null +++ b/http_api_cerberus/README.rst @@ -0,0 +1 @@ +waiting the bot :) diff --git a/http_api_cerberus/__init__.py b/http_api_cerberus/__init__.py new file mode 100644 index 00000000..65b59e22 --- /dev/null +++ b/http_api_cerberus/__init__.py @@ -0,0 +1 @@ +from . import params diff --git a/http_api_cerberus/__manifest__.py b/http_api_cerberus/__manifest__.py new file mode 100644 index 00000000..f3f48cfd --- /dev/null +++ b/http_api_cerberus/__manifest__.py @@ -0,0 +1,15 @@ +# Copyright 2026 Camptocamp SA +# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl). + +{ + "name": "HTTP API Cerberus", + "summary": "Validate HTTP JSON params and results with Cerberus schemas.", + "version": "18.0.1.0.0", + "license": "LGPL-3", + "development_status": "Alpha", + "author": "Camptocamp,Odoo Community Association (OCA)", + "maintainers": ["simahawk"], + "website": "https://github.com/OCA/web-api", + "depends": ["http_api_dispatcher"], + "external_dependencies": {"python": ["cerberus"]}, +} diff --git a/http_api_cerberus/params.py b/http_api_cerberus/params.py new file mode 100644 index 00000000..05760190 --- /dev/null +++ b/http_api_cerberus/params.py @@ -0,0 +1,192 @@ +# Copyright 2018 ACSONE SA/NV +# Copyright 2026 Camptocamp SA +# @author: Simone Orsi +# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl). +"""Cerberus params for `http_api` routes. + +Heavily inspired by `CerberusValidator` and `CerberusListValidator` +in `base_rest.restapi` (OCA/rest-framework): same validation behavior, +same error messages. +""" + +import logging + +from cerberus import Validator + +from odoo.addons.http_api_dispatcher.errors import ApiValidationError +from odoo.addons.http_api_dispatcher.params import ApiParam + +from .tools import cerberus_to_json + +_logger = logging.getLogger(__name__) + + +class CerberusParam(ApiParam): + """Validate and normalize data with a Cerberus schema. + + :param schema: the Cerberus schema, as: + + * a dict + * a `cerberus.Validator` instance (not thread-safe: prefer a dict) + * a callable `(call, direction) -> schema` + * a string: name of a method returning the schema, + looked up on the owner of the endpoint (see `_get_schema_owner`) + + :param purge_unknown: drop keys not declared in the schema + :param validator_cls: `cerberus.Validator` subclass to use + (eg: to provide custom rules or coercers) + :param as_list: data is a list of items matching the schema + :param min_items: minimum number of items (list only) + :param max_items: maximum number of items (list only) + :param unique_items: documentation only (list only) + :param cache: cache the resolved schema (per database and direction). + Use it only when the schema does not depend on the request. + :param strict_output: raise on invalid results. + If False, invalid results are logged and returned as they are. + """ + + def __init__( + self, + schema, + *, + purge_unknown=True, + validator_cls=None, + as_list=False, + min_items=None, + max_items=None, + unique_items=None, + cache=False, + strict_output=True, + ): + """Store the options: the schema is resolved lazily, per call.""" + self.schema = schema + self.purge_unknown = purge_unknown + self.validator_cls = validator_cls or Validator + self.as_list = as_list + self.min_items = min_items + self.max_items = max_items + self.unique_items = unique_items + self.cache = cache + self.strict_output = strict_output + self._schema_cache = {} + + def __repr__(self): + """Show the schema to ease debugging of routes.""" + return f"<{self.__class__.__name__}: {self.schema!r}>" + + # Schema lookup + + def _get_schema_owner(self, call): + """Return the object holding schema methods. + + `call.owner` if available (eg: a service), + otherwise the controller of the endpoint, otherwise `call` itself. + """ + owner = getattr(call, "owner", None) + if owner is not None: + return owner + endpoint = getattr(call, "endpoint", None) + func = getattr(endpoint, "func", endpoint) + return getattr(func, "__self__", call) + + def _resolve_schema(self, call, direction): + """Turn the declared schema into a Cerberus schema or validator.""" + schema = self.schema + if isinstance(schema, str): + schema = getattr(self._get_schema_owner(call), schema)() + elif callable(schema) and not isinstance(schema, Validator): + schema = schema(call, direction) + return schema + + def get_schema(self, call, direction): + """Return the schema, from the cache if enabled. + + The cache is per database: callables might return db dependent schemas. + """ + if not self.cache: + return self._resolve_schema(call, direction) + key = (call.env.cr.dbname, direction) + if key not in self._schema_cache: + self._schema_cache[key] = self._resolve_schema(call, direction) + return self._schema_cache[key] + + def get_validator(self, call, direction): + """Return the validator for given direction. + + Dict schemas get a new validator per call: + validators hold state and are not thread safe. + """ + schema = self.get_schema(call, direction) + if isinstance(schema, Validator): + return schema + if isinstance(schema, dict): + return self.validator_cls(schema, purge_unknown=self.purge_unknown) + raise ValueError(f"Unable to get a Cerberus schema from {self.schema!r}") + + # ApiParam + + def from_params(self, call): + """Validate and normalize the params of the call.""" + params = call.params + if self.as_list: + return self._validate_list(call, params, "input") + validator = self.get_validator(call, "input") + if validator.validate(params): + return validator.document + raise ApiValidationError( + call.env._("BadRequest %s", validator.errors), errors=validator.errors + ) + + def to_response(self, call, result): + """Validate and normalize the result of the endpoint.""" + if self.as_list: + return self._validate_list(call, result, "output") + validator = self.get_validator(call, "output") + if validator.validate(result): + return validator.document + return self._invalid_output(call, result, validator.errors) + + def to_json_schema(self, call, direction): + """Describe the param as JSON schema, for API docs.""" + json_schema = cerberus_to_json(self.get_validator(call, direction).schema) + if not self.as_list: + return json_schema + json_schema = {"type": "array", "items": json_schema} + if self.min_items is not None: + json_schema["minItems"] = self.min_items + if self.max_items is not None: + json_schema["maxItems"] = self.max_items + if self.unique_items is not None: + json_schema["uniqueItems"] = self.unique_items + return json_schema + + # Helpers + + def _invalid_output(self, call, result, errors): + """Raise, or only log when output is not strict.""" + if self.strict_output: + raise SystemError(call.env._("Invalid Response %s", errors)) + _logger.warning("Invalid response (not enforced): %s", errors) + return result + + def _validate_list(self, call, data, direction): + """Validate each item and the size of the list, collecting all errors.""" + validator = self.get_validator(call, direction) + data = data or [] + errors = {} + values = [] + for idx, item in enumerate(data): + if validator.validate(item): + values.append(validator.document) + else: + errors[idx] = validator.errors + count = len(data) + if self.min_items is not None and count < self.min_items: + errors["min_items"] = [f"{count} < {self.min_items}"] + if self.max_items is not None and count > self.max_items: + errors["max_items"] = [f"{count} > {self.max_items}"] + if not errors: + return values + if direction == "output": + return self._invalid_output(call, data, errors) + raise ApiValidationError(call.env._("BadRequest %s", errors), errors=errors) diff --git a/http_api_cerberus/pyproject.toml b/http_api_cerberus/pyproject.toml new file mode 100644 index 00000000..4231d0cc --- /dev/null +++ b/http_api_cerberus/pyproject.toml @@ -0,0 +1,3 @@ +[build-system] +requires = ["whool"] +build-backend = "whool.buildapi" diff --git a/http_api_cerberus/readme/CONTRIBUTORS.md b/http_api_cerberus/readme/CONTRIBUTORS.md new file mode 100644 index 00000000..2b66303a --- /dev/null +++ b/http_api_cerberus/readme/CONTRIBUTORS.md @@ -0,0 +1 @@ +- Simone Orsi \<\> diff --git a/http_api_cerberus/readme/CREDITS.md b/http_api_cerberus/readme/CREDITS.md new file mode 100644 index 00000000..c8995d44 --- /dev/null +++ b/http_api_cerberus/readme/CREDITS.md @@ -0,0 +1,5 @@ +The behavior of the params is derived from `base_rest`'s `restapi.CerberusValidator` +(OCA/rest-framework) by ACSONE SA/NV. + +`cerberus_to_json` and its helpers are heavily inspired by `base_rest.tools` +(OCA/rest-framework) by ACSONE SA/NV. diff --git a/http_api_cerberus/readme/DESCRIPTION.md b/http_api_cerberus/readme/DESCRIPTION.md new file mode 100644 index 00000000..44b22db9 --- /dev/null +++ b/http_api_cerberus/readme/DESCRIPTION.md @@ -0,0 +1,12 @@ +Validate and normalize params and results of `http_api` routes +with [Cerberus](https://docs.python-cerberus.org/) schemas. + +Provides: + +- `CerberusParam`: an `http_api_dispatcher` param validating input params or results +- `cerberus_to_json`: conversion of Cerberus schemas to JSON schemas, + used to generate API docs + +The behavior is compatible with `base_rest`'s `restapi.CerberusValidator`: +unknown keys are dropped, invalid input gives a 400 error and +invalid output gives a 500 error. diff --git a/http_api_cerberus/readme/ROADMAP.md b/http_api_cerberus/readme/ROADMAP.md new file mode 100644 index 00000000..00bf1620 --- /dev/null +++ b/http_api_cerberus/readme/ROADMAP.md @@ -0,0 +1 @@ +- Use the input schema to parse lists from the query string (repeated keys). diff --git a/http_api_cerberus/readme/USAGE.md b/http_api_cerberus/readme/USAGE.md new file mode 100644 index 00000000..5c634d7b --- /dev/null +++ b/http_api_cerberus/readme/USAGE.md @@ -0,0 +1,52 @@ +Declare params on `http_api` routes: + + from odoo import http + from odoo.addons.http_api_cerberus.params import CerberusParam + + class PartnerAPI(http.Controller): + + def _get_search_schema(self): + return { + "name": {"type": "string", "required": True}, + "limit": {"type": "integer", "coerce": int, "default": 10}, + } + + @http.route( + "/api/partner/search", + type="http_api", + auth="user", + methods=["GET"], + input=CerberusParam("_get_search_schema"), + output=CerberusParam( + {"id": {"type": "integer"}, "name": {"type": "string"}}, + as_list=True, + ), + ) + def search(self, name, limit): + partners = http.request.env["res.partner"].search( + [("name", "ilike", name)], limit=limit + ) + return [{"id": p.id, "name": p.name} for p in partners] + +The schema can be: + +- a dict +- a callable `(ctx, direction) -> dict` (`direction` is `input` or `output`) +- the name of a method returning the schema. The method is looked up on + `ctx.owner` if any (eg: a service), otherwise on the controller of the route +- a `cerberus.Validator` instance (not thread safe, prefer the other options) + +Options: + +| Option | Default | Meaning | +|---|---|---| +| `purge_unknown` | `True` | drop keys not declared in the schema | +| `validator_cls` | `cerberus.Validator` | subclass providing custom rules or coercers | +| `as_list` | `False` | data is a list of items matching the schema | +| `min_items`, `max_items` | `None` | bounds for lists | +| `unique_items` | `None` | documentation only | +| `cache` | `False` | cache the resolved schema per database and direction. Use it only when the schema does not depend on the request | +| `strict_output` | `True` | if `False`, invalid results are logged and returned anyway | + +Invalid input raises `ApiValidationError`: +the details of the errors are returned in the `errors` key of the response. diff --git a/http_api_cerberus/static/description/icon.png b/http_api_cerberus/static/description/icon.png new file mode 100644 index 00000000..3a0328b5 Binary files /dev/null and b/http_api_cerberus/static/description/icon.png differ diff --git a/http_api_cerberus/static/description/index.html b/http_api_cerberus/static/description/index.html new file mode 100644 index 00000000..9ac9a659 --- /dev/null +++ b/http_api_cerberus/static/description/index.html @@ -0,0 +1,566 @@ + + + + + +HTTP JSON Cerberus + + + +
+

HTTP JSON Cerberus

+ + +

Alpha License: LGPL-3 OCA/web-api Translate me on Weblate Try me on Runboat

+

Validate and normalize params and results of http_json routes with +Cerberus schemas.

+

Provides:

+
    +
  • CerberusParam: an http_json_dispatcher param validating input +params or results
    • +
  • cerberus_to_json: conversion of Cerberus schemas to JSON schemas, +used to generate API docs
    • +
+

The behavior is compatible with base_rest’s +restapi.CerberusValidator: unknown keys are dropped, invalid input +gives a 400 error and invalid output gives a 500 error.

+
+

Important

+

This is an alpha version, the data model and design can change at any time without warning. +Only for development or testing purpose, do not use in production. +More details on development status

+
+

Table of contents

+ +
+

Usage

+

Declare params on http_json routes:

+
+from odoo import http
+from odoo.addons.http_json_cerberus.params import CerberusParam
+
+class PartnerAPI(http.Controller):
+
+    def _get_search_schema(self):
+        return {
+            "name": {"type": "string", "required": True},
+            "limit": {"type": "integer", "coerce": int, "default": 10},
+        }
+
+    @http.route(
+        "/api/partner/search",
+        type="http_json",
+        auth="user",
+        methods=["GET"],
+        input=CerberusParam("_get_search_schema"),
+        output=CerberusParam(
+            {"id": {"type": "integer"}, "name": {"type": "string"}},
+            as_list=True,
+        ),
+    )
+    def search(self, name, limit):
+        partners = http.request.env["res.partner"].search(
+            [("name", "ilike", name)], limit=limit
+        )
+        return [{"id": p.id, "name": p.name} for p in partners]
+
+

The schema can be:

+
    +
  • a dict
    • +
  • a callable (ctx, direction) -> dict (direction is input or +output)
    • +
  • the name of a method returning the schema. The method is looked up on +ctx.owner if any (eg: a service), otherwise on the controller of +the route
    • +
  • a cerberus.Validator instance (not thread safe, prefer the other +options)
    • +
+

Options:

+ +++++ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
OptionDefaultMeaning
purge_unknownTruedrop keys not +declared in the +schema
validator_cls`` +cerberus.Validator``subclass providing +custom rules or +coercers
as_listFalsedata is a list of +items matching the +schema
min_items, +max_itemsNonebounds for lists
unique_itemsNonedocumentation only
cacheFalsecache the resolved +schema per database +and direction. Use +it only when the +schema does not +depend on the +request
strict_outputTrueif False, +invalid results are +logged and returned +anyway
+

Invalid input raises ApiValidationError: the details of the errors +are returned in the errors key of the response.

+
+
+

Known issues / Roadmap

+
    +
  • Use the input schema to parse lists from the query string (repeated +keys).
    • +
+
+
+

Bug Tracker

+

Bugs are tracked on GitHub Issues. +In case of trouble, please check there if your issue has already been reported. +If you spotted it first, help us to smash it by providing a detailed and welcomed +feedback.

+

Do not contact contributors directly about support or help with technical issues.

+
+
+

Credits

+
+

Authors

+
    +
  • Camptocamp
    • +
+
+ +
+

Other credits

+

The behavior of the params is derived from base_rest’s +restapi.CerberusValidator (OCA/rest-framework) by ACSONE SA/NV.

+

cerberus_to_json and its helpers are heavily inspired by +base_rest.tools (OCA/rest-framework) by ACSONE SA/NV.

+
+
+

Maintainers

+

This module is maintained by the OCA.

+ +Odoo Community Association + +

OCA, or the Odoo Community Association, is a nonprofit organization whose +mission is to support the collaborative development of Odoo features and +promote its widespread use.

+

Current maintainer:

+

simahawk

+

This module is part of the OCA/web-api project on GitHub.

+

You are welcome to contribute. To learn how please visit https://odoo-community.org/page/Contribute.

+
+
+
+ + diff --git a/http_api_cerberus/tests/__init__.py b/http_api_cerberus/tests/__init__.py new file mode 100644 index 00000000..dc1ace64 --- /dev/null +++ b/http_api_cerberus/tests/__init__.py @@ -0,0 +1,3 @@ +from . import test_params +from . import test_tools +from . import test_http diff --git a/http_api_cerberus/tests/test_http.py b/http_api_cerberus/tests/test_http.py new file mode 100644 index 00000000..34e30c06 --- /dev/null +++ b/http_api_cerberus/tests/test_http.py @@ -0,0 +1,55 @@ +# Copyright 2026 Camptocamp SA +# @author: Simone Orsi +# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl). + +import json + +from odoo import http +from odoo.tests.common import HttpCase, tagged +from odoo.tools import mute_logger + +from ..params import CerberusParam + + +class CerberusTestController(http.Controller): + def _get_input_schema(self): + return {"name": {"type": "string", "required": True}} + + @http.route( + "/http_api_cerberus_test/hello", + type="http_api", + auth="public", + csrf=False, + methods=["POST"], + input=CerberusParam("_get_input_schema"), + output=CerberusParam({"greeting": {"type": "string", "required": True}}), + ) + def hello(self, name): + return {"greeting": f"Hello {name}", "secret": "dropped"} + + +@tagged("-at_install", "post_install") +class TestHttp(HttpCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.env.registry.clear_cache("routing") + + def _post(self, data): + return self.url_open( + "/http_api_cerberus_test/hello", + data=json.dumps(data), + headers={"Content-Type": "application/json"}, + ) + + def test_ok(self): + """A valid request is served; keys outside the schemas are dropped.""" + response = self._post({"name": "John", "other": 1}) + self.assertEqual(response.json(), {"greeting": "Hello John"}) + + @mute_logger("odoo.addons.http_api_dispatcher.errors", "odoo.http") + def test_bad_request(self): + """Invalid input gives a 400 w/ the Cerberus errors.""" + response = self._post({}) + self.assertEqual(response.status_code, 400) + self.assertEqual(response.json()["errors"], {"name": ["required field"]}) diff --git a/http_api_cerberus/tests/test_params.py b/http_api_cerberus/tests/test_params.py new file mode 100644 index 00000000..ad9ef638 --- /dev/null +++ b/http_api_cerberus/tests/test_params.py @@ -0,0 +1,134 @@ +# Copyright 2026 Camptocamp SA +# @author: Simone Orsi +# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl). + +from types import SimpleNamespace + +from cerberus import Validator + +from odoo.tests.common import TransactionCase +from odoo.tools import mute_logger + +from odoo.addons.http_api_dispatcher.errors import ApiValidationError + +from ..params import CerberusParam + +SCHEMA = { + "name": {"type": "string", "required": True}, + "qty": {"type": "integer", "coerce": int}, +} + + +class Owner: + calls = 0 + + def _get_schema(self): + Owner.calls += 1 + return SCHEMA + + +class TestCerberusParam(TransactionCase): + def _call(self, params=None, owner=None): + """Minimal stand-in for an `ApiCall`: env, params and an owner.""" + return SimpleNamespace(env=self.env, owner=owner, params=params or {}) + + def test_input(self): + """Input is validated and coerced; unknown keys are dropped.""" + param = CerberusParam(SCHEMA) + res = param.from_params(self._call({"name": "a", "qty": "2", "x": 1})) + self.assertEqual(res, {"name": "a", "qty": 2}) + + def test_input_error(self): + """Missing required keys raise an error w/ the Cerberus errors.""" + param = CerberusParam(SCHEMA) + with self.assertRaises(ApiValidationError) as err: + param.from_params(self._call({"qty": 1})) + self.assertEqual( + err.exception.rest_json_info, {"errors": {"name": ["required field"]}} + ) + self.assertIn("BadRequest", str(err.exception)) + + def test_keep_unknown(self): + """With `purge_unknown=False`, unknown keys are an error.""" + param = CerberusParam(SCHEMA, purge_unknown=False) + with self.assertRaises(ApiValidationError): + param.from_params(self._call({"name": "a", "x": 1})) + + def test_output(self): + """Output follows the schema; invalid output raises a SystemError.""" + param = CerberusParam(SCHEMA) + self.assertEqual(param.to_response(self._call(), {"name": "a"}), {"name": "a"}) + with self.assertRaises(SystemError): + param.to_response(self._call(), {"qty": 1}) + + @mute_logger("odoo.addons.http_api_cerberus.params") + def test_output_not_strict(self): + """With `strict_output=False`, invalid output is returned as is.""" + param = CerberusParam(SCHEMA, strict_output=False) + self.assertEqual(param.to_response(self._call(), {"qty": 1}), {"qty": 1}) + + def test_schema_lookup(self): + """The schema can be a method name, a callable or a Validator.""" + Owner.calls = 0 + param = CerberusParam("_get_schema") + owner = Owner() + self.assertEqual( + param.from_params(self._call({"name": "a"}, owner)), {"name": "a"} + ) + param.from_params(self._call({"name": "a"}, owner)) + self.assertEqual(Owner.calls, 2) + param = CerberusParam(lambda call, direction: SCHEMA) + self.assertEqual( + param.from_params(self._call({"name": "a"}, owner)), {"name": "a"} + ) + param = CerberusParam(Validator(SCHEMA, purge_unknown=True)) + self.assertEqual( + param.from_params(self._call({"name": "a"}, owner)), {"name": "a"} + ) + + def test_schema_cache(self): + """With `cache=True`, the schema method is looked up once per direction.""" + Owner.calls = 0 + param = CerberusParam("_get_schema", cache=True) + owner = Owner() + param.from_params(self._call({"name": "a"}, owner)) + param.from_params(self._call({"name": "b"}, owner)) + self.assertEqual(Owner.calls, 1) + param.to_response(self._call(owner=owner), {"name": "b"}) + self.assertEqual(Owner.calls, 2) + + def test_list(self): + """`as_list` validates each item and the min/max number of items.""" + param = CerberusParam(SCHEMA, as_list=True, min_items=1, max_items=2) + self.assertEqual( + param.to_response(self._call(), [{"name": "a", "x": 1}, {"name": "b"}]), + [{"name": "a"}, {"name": "b"}], + ) + with self.assertRaises(SystemError): + param.to_response(self._call(), []) + with self.assertRaises(ApiValidationError) as err: + param.from_params(self._call([{"name": "a"}, {"qty": 1}, {"name": "c"}])) + self.assertEqual( + err.exception.rest_json_info["errors"], + {1: {"name": ["required field"]}, "max_items": ["3 > 2"]}, + ) + + def test_json_schema(self): + """The schema is converted to the JSON schema of a list of objects.""" + param = CerberusParam(SCHEMA, as_list=True, min_items=1, unique_items=True) + self.assertEqual( + param.to_json_schema(self._call(), "output"), + { + "type": "array", + "minItems": 1, + "uniqueItems": True, + "items": { + "type": "object", + "required": ["name"], + "properties": { + "name": {"type": "string"}, + "qty": {"type": "integer"}, + }, + }, + }, + ) diff --git a/http_api_cerberus/tests/test_tools.py b/http_api_cerberus/tests/test_tools.py new file mode 100644 index 00000000..9db337e0 --- /dev/null +++ b/http_api_cerberus/tests/test_tools.py @@ -0,0 +1,79 @@ +# Copyright 2026 Camptocamp SA +# @author: Simone Orsi +# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl). + +from odoo.tests.common import BaseCase + +from ..tools import cerberus_to_json + + +class TestTools(BaseCase): + def test_cerberus_to_json(self): + """Each Cerberus rule maps to its JSON schema equivalent.""" + schema = { + "name": { + "type": "string", + "required": True, + "minlength": 1, + "maxlength": 10, + "meta": {"title": "Name", "example": "John"}, + }, + "kind": {"type": "string", "allowed": ["a", "b"], "default": "a"}, + "qty": {"type": "float", "min": 0, "max": 5, "nullable": True}, + "when": {"type": "datetime", "readonly": True}, + "day": {"type": "date"}, + "code": {"type": "string", "regex": "^[A-Z]+$"}, + "ids": {"type": "list", "schema": {"type": "integer"}, "minlength": 1}, + "tags": {"type": "list"}, + "child": { + "type": "dict", + "schema": {"id": {"type": "integer", "required": True}}, + }, + "mapping": {"type": "dict", "valuesrules": {"type": "integer"}}, + "any": {"type": ["string", "integer"]}, + "untyped": {"required": True}, + } + self.assertEqual( + cerberus_to_json(schema), + { + "type": "object", + "required": ["name", "untyped"], + "properties": { + "name": { + "type": "string", + "minLength": 1, + "maxLength": 10, + "title": "Name", + "example": "John", + }, + "kind": {"type": "string", "enum": ["a", "b"], "default": "a"}, + "qty": { + "type": "number", + "format": "float", + "minimum": 0, + "maximum": 5, + "nullable": True, + }, + "when": {"type": "string", "format": "date-time", "readOnly": True}, + "day": {"type": "string", "format": "date"}, + "code": {"type": "string", "pattern": "^[A-Z]+$"}, + "ids": { + "type": "array", + "minItems": 1, + "items": {"type": "integer"}, + }, + "tags": {"type": "array", "items": {"type": "string"}}, + "child": { + "type": "object", + "required": ["id"], + "properties": {"id": {"type": "integer"}}, + }, + "mapping": { + "type": "object", + "additionalProperties": {"type": "integer"}, + }, + "any": {"anyOf": [{"type": "string"}, {"type": "integer"}]}, + "untyped": {}, + }, + }, + ) diff --git a/http_api_cerberus/tools.py b/http_api_cerberus/tools.py new file mode 100644 index 00000000..24774d19 --- /dev/null +++ b/http_api_cerberus/tools.py @@ -0,0 +1,111 @@ +# Copyright 2018 ACSONE SA/NV +# Copyright 2026 Camptocamp SA +# @author: Simone Orsi +# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl). +"""Convert Cerberus schemas to JSON schemas (OpenAPI flavor). + +Heavily inspired by `cerberus_to_json` and its helpers in `base_rest.tools` +(OCA/rest-framework, ACSONE SA/NV): same mapping of rules, +plus support for `date` and multiple types. +""" + +import logging + +_logger = logging.getLogger(__name__) + +# Cerberus type -> (JSON type, JSON format) +TYPE_MAP = { + "boolean": ("boolean", None), + "date": ("string", "date"), + "datetime": ("string", "date-time"), + "dict": ("object", None), + "float": ("number", "float"), + "integer": ("integer", None), + "list": ("array", None), + "number": ("number", None), + "objectid": ("string", "objectid"), + "string": ("string", None), +} +# Cerberus rule -> JSON keyword, for rules w/ the same value in both +SIMPLE_RULES = { + "allowed": "enum", + "default": "default", + "description": "description", + "nullable": "nullable", + "readonly": "readOnly", + "regex": "pattern", +} +SUPPORTED_META = ("title", "description", "example", "examples") +NUMERIC_TYPES = ("integer", "float", "number") + + +def cerberus_to_json(schema): + """Convert a Cerberus schema (a dict of fields) to a JSON object schema.""" + return { + "type": "object", + # sorted to get stable docs + "required": sorted(k for k, spec in schema.items() if spec.get("required")), + "properties": {k: field_to_json(spec) for k, spec in schema.items()}, + } + + +def field_to_json(spec): + """Convert the Cerberus rules of a field to a JSON schema.""" + _type = spec.get("type") + if _type is None: + return {} + if isinstance(_type, list | tuple): + return {"anyOf": [field_to_json(dict(spec, type=t)) for t in _type]} + json_type, json_format = TYPE_MAP.get(_type, (_type, None)) + res = {"type": json_type} + if json_format: + res["format"] = json_format + for rule, keyword in SIMPLE_RULES.items(): + if rule in spec: + res[keyword] = spec[rule] + for key in SUPPORTED_META: + value = spec.get("meta", {}).get(key) + if value: + res[key] = value + _add_bounds(_type, spec, res) + if json_type == "object": + _add_object_rules(spec, res) + elif json_type == "array": + res["items"] = field_to_json(spec["schema"]) if "schema" in spec else {} + if not res["items"]: + res["items"] = {"type": "string"} + return res + + +def _add_bounds(_type, spec, res): + """Map min/max rules: lengths for strings and lists, values for numbers.""" + length_keys = { + "string": ("minLength", "maxLength"), + "list": ("minItems", "maxItems"), + } + if _type in length_keys: + min_key, max_key = length_keys[_type] + if "minlength" in spec: + res[min_key] = spec["minlength"] + if "maxlength" in spec: + res[max_key] = spec["maxlength"] + if _type in NUMERIC_TYPES: + if "min" in spec: + res["minimum"] = spec["min"] + if "max" in spec: + res["maximum"] = spec["max"] + + +def _add_object_rules(spec, res): + """Map nested schemas and `valuesrules` of dicts.""" + if "schema" in spec: + res.update(cerberus_to_json(spec["schema"])) + keys_type = spec.get("keysrules", {}).get("type", "string") + if keys_type != "string": + _logger.debug("JSON schema supports only string keys. Got: %s", keys_type) + if "valuesrules" in spec: + values_rules = spec["valuesrules"] + additional = {"type": TYPE_MAP.get(values_rules.get("type"), ("string",))[0]} + if "schema" in values_rules: + additional.update(cerberus_to_json(values_rules["schema"])) + res["additionalProperties"] = additional diff --git a/test-requirements.txt b/test-requirements.txt new file mode 100644 index 00000000..9838a683 --- /dev/null +++ b/test-requirements.txt @@ -0,0 +1 @@ +odoo-addon-http_api_dispatcher @ git+https://github.com/OCA/web-api.git@refs/pull/179/head#subdirectory=http_api_dispatcher