Merge pull request #6 from Fatal1ty/plugins

Add plugins functionality

.github/workflows/main.yml

@@ -14,7 +14,7 @@ jobs:
  runs-on: ubuntu-latest
  strategy:
  matrix:
- python-version: ["3.7", "3.8", "3.9", "3.10", "3.11", "3.12"]
+ python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
  steps:
  - uses: actions/checkout@v3
  - name: Set up Python ${{ matrix.python-version }}
@@ -27,7 +27,7 @@ jobs:
  pip install .
  pip install -r requirements-dev.txt
  - name: Run flake8
- run: flake8 openapify --ignore=E203,W503
+ run: flake8 openapify --ignore=E203,W503,E704
  - name: Run mypy
  run: mypy openapify
  - name: Run black

README.md

@@ -42,7 +42,9 @@ Table of contents
  * [Request](#request)
  * [Response](#response)
  * [Security requirements](#security-requirements)
-* [Entity schema builders](#entity-schema-builders)
+* [Plugins](#plugins)
+ * [`schema_helper`](#schema_helper)
+ * [`media_type_helper`](#media_type_helper)
 
 Installation
 --------------------------------------------------------------------------------
@@ -1047,13 +1049,127 @@ components:
  in: header
 ```
 
-Entity schema builders
+Plugins
 --------------------------------------------------------------------------------
 
-In some decorators you should pass Python data type for which the JSON Schema
-is being built by openapify in order to get the correct OpenAPI document.
-Out of the box, the schema is generated by
-using [`mashumaro`](https://github.com/Fatal1ty/mashumaro) library, but support
-for third-party entity schema generators can be implemented through an apispec
-plugin. In the future, this chapter will contain recommendations for writing
-and using such plugins.
+Some aspects of creating an OpenAPI document can be changed using plugins.
+There is `openapify.plugins.BasePlugin` base class, which has all the methods
+available for definition. If you want to write a plugin that, for example, will
+only generate schema for request parameters, then it will be enough for you to
+define only one appropriate method, and leave the rest non-implemented.
+Plugin system works by going through all registered plugins and calling
+the appropriate method. If such a method raises `NotImplementedError` or
+returns `None`, it is assumed that this plugin doesn't provide the necessary
+functionality. Iteration stops at the first plugin that returned something
+other than `None`.
+
+Plugins are registered via the `plugins` argument of the `build_spec` function:
+
+```python
+from openapify import BasePlugin, build_spec
+
+
+class MyPlugin1(BasePlugin):
+ def schema_helper(...):
+ # return something meaningful here, see the following chapters
+ ...
+
+
+build_spec(..., plugins=[MyPlugin1()])
+```
+
+### `schema_helper`
+
+OpenAPI [Schema](https://spec.openapis.org/oas/v3.1.0#schemaObject) object
+is built from python types stored in the `value_type` attribute of the
+following openapify dataclasses defined in `openapify.core.models`:
+* `Body`
+* `Cookie`
+* `Header`
+* `QueryParam`
+
+Out of the box, the schema is generated by using
+[`mashumaro`](https://github.com/Fatal1ty/mashumaro) library, but support
+for third-party entity schema generators can be achieved through
+`schema_helper` method. For example, here's what a plugin for pydantic models
+might look like:
+
+```python
+from typing import Any
+
+from openapify import BasePlugin
+from openapify.core.models import Body, Cookie, Header, QueryParam
+from pydantic import BaseModel
+
+
+class PydanticSchemaPlugin(BasePlugin):
+ def schema_helper(
+ self,
+ obj: Body | Cookie | Header | QueryParam,
+ name: str | None = None,
+ ) -> dict[str, Any] | None:
+ if issubclass(obj.value_type, BaseModel):
+ schema = obj.value_type.model_json_schema(
+ ref_template="#/components/schemas/{model}"
+ )
+ self.spec.components.schemas.update(schema.pop("$defs", {}))
+ return schema
+```
+
+### media_type_helper
+
+A media type is used in OpenAPI Request
+[Body](https://spec.openapis.org/oas/v3.1.0#request-body-object) and
+[Response](https://spec.openapis.org/oas/v3.1.0#response-object) objects.
+By default, `application/octet-stream` is applied for `bytes` or `bytearray`
+types, and `application/json` is applied otherwise. You can support mode media
+types or override existing ones with `media_type_helper` method.
+
+Let's imagine that you have an API route that returns PNG images as the body.
+You can have a separate model class representing images, but the more common
+case is to use `typing.Annotated` wrapper for bytes. Here's what a plugin for
+`image/png` media type might look like:
+
+```python
+from typing import Annotated, Any, Dict, Optional
+
+from openapify import BasePlugin, build_spec, response_schema
+from openapify.core.models import Body, RouteDef
+
+ImagePNG = Annotated[bytes, "PNG"]
+
+
+class ImagePNGPlugin(BasePlugin):
+ def media_type_helper(
+ self, body: Body, schema: Dict[str, Any]
+ ) -> Optional[str]:
+ if body.value_type is ImagePNG:
+ return "image/png"
+
+
+@response_schema(body=ImagePNG)
+def foo():
+ ...
+
+
+routes = [RouteDef("/foo", "get", foo)]
+spec = build_spec(routes, plugins=[ImagePNGPlugin()])
+print(spec.to_yaml())
+```
+
+The resulting document will contain `image/png` content in the response:
+```yaml
+openapi: 3.1.0
+info:
+ title: API
+ version: 1.0.0
+paths:
+ /foo:
+ get:
+ responses:
+ '200':
+ description: OK
+ content:
+ image/png:
+ schema: {}
+```

openapify/__init__.py

@@ -8,6 +8,7 @@
  response_schema,
  security_requirements,
 )
+from .plugin import BasePlugin
 
 __all__ = [
  "build_spec",
@@ -20,4 +21,5 @@
  "Header",
  "QueryParam",
  "Example",
+ "BasePlugin",
 ]

openapify/core/base_plugins.py

@@ -0,0 +1,57 @@
+from typing import Any, Dict, Optional, Union
+
+from mashumaro.jsonschema import OPEN_API_3_1, JSONSchemaBuilder
+
+from openapify.core.models import Body, Cookie, Header, QueryParam
+from openapify.core.utils import get_value_type
+from openapify.plugin import BasePlugin
+
+
+class BodyBinaryPlugin(BasePlugin):
+ def schema_helper(
+ self,
+ obj: Union[Body, Cookie, Header, QueryParam],
+ name: Optional[str] = None,
+ ) -> Optional[Dict[str, Any]]:
+ try:
+ if isinstance(obj, Body):
+ if get_value_type(obj.value_type) in (bytes, bytearray):
+ return {}
+
+ return None
+ except TypeError:
+ return None
+
+
+class GuessMediaTypePlugin(BasePlugin):
+ def media_type_helper(
+ self, body: Body, schema: Dict[str, Any]
+ ) -> Optional[str]:
+ if not schema and get_value_type(body.value_type) in (
+ bytes,
+ bytearray,
+ ):
+ return "application/octet-stream"
+ else:
+ return "application/json"
+
+
+class BaseSchemaPlugin(BasePlugin):
+ def schema_helper(
+ self,
+ obj: Union[Body, Cookie, Header, QueryParam],
+ name: Optional[str] = None,
+ ) -> Optional[Dict[str, Any]]:
+ builder = JSONSchemaBuilder(
+ dialect=OPEN_API_3_1, ref_prefix="#/components/schemas"
+ )
+ try:
+ json_schema = builder.build(obj.value_type)
+ except Exception:
+ return None
+ schemas = self.spec.components.schemas
+ for name, schema in builder.context.definitions.items():
+ schemas[name] = schema.to_dict()
+ if isinstance(obj, QueryParam) and obj.default is not None:
+ json_schema.default = obj.default
+ return json_schema.to_dict()