import inspect
import json
import os
import warnings
from functools import lru_cache, partial
from pathlib import Path
from typing import (
Callable,
cast,
Deque,
Dict,
List,
overload,
Tuple,
Type,
Union,
)
import attr
import yaml
from aiohttp import hdrs, web
from aiohttp_middlewares import cors_middleware
from openapi_core.schema.specs.models import Spec
from openapi_core.shortcuts import create_spec
from pyrsistent import pmap
from yarl import URL
from rororo.annotations import (
DictStrAny,
DictStrStr,
F,
Handler,
Protocol,
ViewType,
)
from rororo.openapi import views
from rororo.openapi.annotations import (
CorsMiddlewareKwargsDict,
ErrorMiddlewareKwargsDict,
SecurityDict,
ValidateEmailKwargsDict,
)
from rororo.openapi.constants import (
APP_OPENAPI_SCHEMA_KEY,
APP_OPENAPI_SPEC_KEY,
APP_VALIDATE_EMAIL_KWARGS_KEY,
HANDLER_OPENAPI_MAPPING_KEY,
)
from rororo.openapi.core_data import get_core_operation
from rororo.openapi.exceptions import ConfigurationError
from rororo.openapi.middlewares import openapi_middleware
from rororo.openapi.utils import add_prefix
from rororo.settings import APP_SETTINGS_KEY, BaseSettings
SchemaLoader = Callable[[bytes], DictStrAny]
Url = Union[str, URL]
class CreateSchemaAndSpec(Protocol):
def __call__(
self, path: Path, *, schema_loader: Union[SchemaLoader, None] = None
) -> Tuple[DictStrAny, Spec]: # pragma: no cover
...
[docs]@attr.dataclass(slots=True)
class OperationTableDef:
"""Map OpenAPI 3 operations to aiohttp.web view handlers.
In short it is rororo's equialent to :class:`aiohttp.web.RouteTableDef`.
Under the hood, on :func:`rororo.openapi.setup_openapi` it
still will use ``RouteTableDef`` for registering view handlers to
:class:`aiohttp.web.Application`.
But unlike ``RouteTableDef`` it does not register any HTTP method handlers
(as via ``@routes.get`` decorator) in favor of just registering the
operations.
There are two ways for registering view hanlder,
1. With bare ``@operations.register`` call when OpenAPI ``operationId``
equals to view handler name.
2. Or by passing ``operation_id`` to the decorator as first arg, when
``operationId`` does not match view handler name, or if you don't like
the fact of guessing operation ID from view handler name.
Both of ways described below,
.. code-block:: python
from rororo import OperationTableDef
operations = OperationTableDef()
# Expect OpenAPI 3 schema to contain operationId: hello_world
@operations.register
async def hello_world(request: web.Request) -> web.Response:
...
# Explicitly use `operationId: "helloWorld"`
@operations.register("helloWorld")
async def hello_world(request: web.Request) -> web.Response:
...
Class based views supported as well. In most generic way you just need
to decorate your view with ``@operations.register`` decorator and ensure
that ``operationId`` equals to view method qualified name
(``__qualname__``).
For example,
.. code-block:: python
@operations.register
class UserView(web.View):
async def get(self) -> web.Response:
...
expects for operation ID ``UserView.get`` to be declared in OpenAPI schema.
In same time,
.. code-block:: python
@operations.register("users")
class UserView(web.View):
async def get(self) -> web.Response:
...
expects for operation ID ``users.get`` to be declared in OpenAPI schema.
Finally,
.. code-block:: python
@operations.register
class UserView(web.View):
@operations.register("me")
async def get(self) -> web.Response:
...
expects for operation ID ``me`` to be declared in OpenAPI schema.
When the class based view provides mutliple view methods (for example
``delete``, ``get``, ``patch`` & ``put``) *rororo* expects that
OpenAPI schema contains operation IDs for each of view method.
If supplied ``operation_id`` does not exist in OpenAPI 3 schema,
:func:`rororo.openapi.setup_openapi` call raises an ``OperationError``.
"""
handlers: List[Handler] = attr.Factory(list)
views: List[ViewType] = attr.Factory(list)
def __add__(self, other: "OperationTableDef") -> "OperationTableDef":
return OperationTableDef(
handlers=[*self.handlers, *other.handlers],
views=[*self.views, *other.views],
)
def __iadd__(self, other: "OperationTableDef") -> "OperationTableDef":
self.handlers.extend(other.handlers)
self.views.extend(other.views)
return self
@overload
def register(self, handler: F) -> F:
...
@overload
def register(self, operation_id: str) -> Callable[[F], F]:
...
def register(self, mixed): # type: ignore[no-untyped-def]
operation_id = mixed if isinstance(mixed, str) else mixed.__qualname__
def decorator(handler: F) -> F:
mapping: DictStrStr = {}
if self._is_view(handler):
mapping.update(
self._register_view(handler, operation_id) # type: ignore[arg-type]
)
else:
mapping.update(self._register_handler(handler, operation_id))
setattr(handler, HANDLER_OPENAPI_MAPPING_KEY, pmap(mapping))
return handler
return decorator(mixed) if callable(mixed) else decorator
def _is_view(self, handler: F) -> bool:
return inspect.isclass(handler) and issubclass(handler, web.View)
def _register_handler(
self, handler: Handler, operation_id: str
) -> DictStrStr:
# Hacky way to check whether handler is a view function or view method
has_self_parameter = "self" in inspect.signature(handler).parameters
# Register only view functions, view methods will be registered via
# view class instead
if not has_self_parameter:
self.handlers.append(handler)
return {hdrs.METH_ANY: operation_id}
def _register_view(self, view: ViewType, prefix: str) -> DictStrStr:
mapping: DictStrStr = {}
for value in vars(view).values():
if not callable(value):
continue
name = value.__name__
maybe_method = name.upper()
if maybe_method not in hdrs.METH_ALL:
continue
maybe_operation_id = getattr(
value, HANDLER_OPENAPI_MAPPING_KEY, {}
).get(hdrs.METH_ANY)
mapping[maybe_method] = (
maybe_operation_id
if maybe_operation_id
else f"{prefix}.{name}"
)
self.views.append(view)
return mapping
def convert_operations_to_routes(
operations: OperationTableDef,
spec: Spec,
*,
prefix: Union[str, None] = None,
) -> web.RouteTableDef:
"""Convert operations table defintion to routes table definition."""
async def noop(request: web.Request) -> web.Response:
return web.json_response(status=204) # pragma: no cover
routes = web.RouteTableDef()
# Add plain handlers to the route table def as a route
for handler in operations.handlers:
operation_id = getattr(handler, HANDLER_OPENAPI_MAPPING_KEY)[
hdrs.METH_ANY
]
core_operation = get_core_operation(spec, operation_id)
routes.route(
core_operation.http_method,
add_prefix(core_operation.path_name, prefix),
name=get_route_name(core_operation.operation_id),
)(handler)
# But view should be added as a view instead
for view in operations.views:
ids: Deque[str] = Deque(
getattr(view, HANDLER_OPENAPI_MAPPING_KEY).values()
)
first_operation_id = ids.popleft()
core_operation = get_core_operation(spec, first_operation_id)
path = add_prefix(core_operation.path_name, prefix)
routes.view(
path,
name=get_route_name(core_operation.operation_id),
)(view)
# Hacky way of adding aliases to class based views with multiple
# registered view methods
for other_operation_id in ids:
routes.route(
hdrs.METH_ANY, path, name=get_route_name(other_operation_id)
)(noop)
return routes
def create_schema_and_spec(
path: Path, *, schema_loader: Union[SchemaLoader, None] = None
) -> Tuple[DictStrAny, Spec]:
schema = read_openapi_schema(path, loader=schema_loader)
return (schema, create_spec(schema))
@lru_cache(maxsize=128)
def create_schema_and_spec_with_cache( # type: ignore[misc]
path: Path, *, schema_loader: Union[SchemaLoader, None] = None
) -> Tuple[DictStrAny, Spec]:
return create_schema_and_spec(path, schema_loader=schema_loader)
def find_route_prefix(
oas: DictStrAny,
*,
server_url: Union[Url, None] = None,
settings: Union[BaseSettings, None] = None,
) -> str:
if server_url is not None:
return get_route_prefix(server_url)
servers = oas["servers"]
if len(servers) == 1:
return get_route_prefix(servers[0]["url"])
if settings is None:
raise ConfigurationError(
"Unable to guess route prefix as OpenAPI schema contains "
"multiple servers and aiohttp.web has no settings instance "
"configured."
)
for server in servers:
mixed = server.get("x-rororo-level")
if isinstance(mixed, list):
if settings.level in mixed:
return get_route_prefix(server["url"])
elif mixed == settings.level:
return get_route_prefix(server["url"])
raise ConfigurationError(
"Unable to guess route prefix as no server in OpenAPI schema has "
f'defined "x-rororo-level" key of "{settings.level}".'
)
def fix_spec_operations(spec: Spec, schema: DictStrAny) -> Spec:
"""Fix spec operations.
``openapi-core`` sets up operation security to an empty list even it is
not defined within the operation schema. This function fixes this behaviour
by reading schema first and if operation schema misses ``security``
definition - sets up a ``None`` as an operation security.
This allows properly distinct empty operation security and missed operation
security. With empty operation security (empty list) - mark an operation
as unsecured. With missed operation security - use global security schema
if it is defined.
"""
mapping: Dict[str, Union[SecurityDict, None]] = {}
for path_data in schema["paths"].values():
for maybe_operation_data in path_data.values():
if not isinstance(maybe_operation_data, dict):
continue
operation_id = maybe_operation_data.get("operationId")
if operation_id is None:
continue
mapping[operation_id] = maybe_operation_data.get("security")
for path in spec.paths.values():
for operation in path.operations.values():
if operation.security != []:
continue
if operation.operation_id is None:
continue
operation.security = mapping[operation.operation_id]
return spec
def get_default_yaml_loader() -> Type[yaml.BaseLoader]:
return cast(
Type[yaml.BaseLoader], getattr(yaml, "CSafeLoader", yaml.SafeLoader)
)
def get_route_name(operation_id: str) -> str:
return operation_id.replace(" ", "-")
def get_route_prefix(mixed: Url) -> str:
return (URL(mixed) if isinstance(mixed, str) else mixed).path
[docs]def read_openapi_schema(
path: Path, *, loader: Union[SchemaLoader, None] = None
) -> DictStrAny:
"""Read OpenAPI Schema from given path.
By default, when ``loader`` is not explicitly passed, attempt to guess
schema loader function from path extension.
``loader`` should be a callable, which receives ``bytes`` and returns
``Dict[str, Any]`` of OpenAPI Schema.
By default, next schema loader used,
- :func:`json.loads` for ``openapi.json``
- ``yaml.load`` for ``openapi.yaml``
"""
if loader is None:
if path.suffix == ".json":
loader = json.loads
elif path.suffix in {".yml", ".yaml"}:
loader = partial(yaml.load, Loader=get_default_yaml_loader())
if loader is not None:
return loader(path.read_bytes())
raise ConfigurationError(
f"Unsupported OpenAPI schema file: {path}. At a moment rororo "
"supports loading OpenAPI schemas from: .json, .yml, .yaml files"
)
@overload
def setup_openapi(
app: web.Application,
schema_path: Union[str, Path],
*operations: OperationTableDef,
server_url: Union[Url, None] = None,
is_validate_response: bool = True,
has_openapi_schema_handler: bool = True,
use_error_middleware: bool = True,
error_middleware_kwargs: Union[ErrorMiddlewareKwargsDict, None] = None,
use_cors_middleware: bool = True,
cors_middleware_kwargs: Union[CorsMiddlewareKwargsDict, None] = None,
schema_loader: Union[SchemaLoader, None] = None,
cache_create_schema_and_spec: bool = False,
validate_email_kwargs: Union[ValidateEmailKwargsDict, None] = None,
) -> web.Application:
...
@overload
def setup_openapi(
app: web.Application,
*operations: OperationTableDef,
schema: DictStrAny,
spec: Spec,
server_url: Union[Url, None] = None,
is_validate_response: bool = True,
has_openapi_schema_handler: bool = True,
use_error_middleware: bool = True,
error_middleware_kwargs: Union[ErrorMiddlewareKwargsDict, None] = None,
use_cors_middleware: bool = True,
cors_middleware_kwargs: Union[CorsMiddlewareKwargsDict, None] = None,
validate_email_kwargs: Union[ValidateEmailKwargsDict, None] = None,
) -> web.Application:
...
[docs]def setup_openapi( # type: ignore[misc]
app: web.Application,
schema_path: Union[str, Path, None] = None,
*operations: OperationTableDef,
schema: Union[DictStrAny, None] = None,
spec: Union[Spec, None] = None,
server_url: Union[Url, None] = None,
is_validate_response: bool = True,
has_openapi_schema_handler: bool = True,
use_error_middleware: bool = True,
error_middleware_kwargs: Union[ErrorMiddlewareKwargsDict, None] = None,
use_cors_middleware: bool = True,
cors_middleware_kwargs: Union[CorsMiddlewareKwargsDict, None] = None,
schema_loader: Union[SchemaLoader, None] = None,
cache_create_schema_and_spec: bool = False,
validate_email_kwargs: Union[ValidateEmailKwargsDict, None] = None,
) -> web.Application:
"""Setup OpenAPI schema to use with aiohttp.web application.
Unlike `aiohttp-apispec <https://aiohttp-apispec.readthedocs.io/>`_ and
other tools, which provides OpenAPI/Swagger support for aiohttp.web
applications, *rororo* changes the way of using OpenAPI schema with
``aiohttp.web`` apps.
*rororo* using schema first approach and relies on concrete OpenAPI schema
file, path to which need to be registered on application startup (mostly
inside of ``create_app`` factory or right after
:class:`aiohttp.web.Application` instantiation).
And as valid OpenAPI schema ensure unique ``operationId`` used accross the
schema *rororo* uses them as a key while telling aiohttp.web to use given
view handler for serving required operation.
With that in mind registering (setting up) OpenAPI schema requires:
1. :class:`aiohttp.web.Application` instance
2. Path to file (json or yaml) with OpenAPI schema
3. OpenAPI operation handlers mapping (rororo's equialent of
:class:`aiohttp.web.RouteTableDef`)
In common cases setup looks like,
.. code-block:: python
from pathlib import Path
from typing import List
from aiohttp import web
from .views import operations
def create_app(argv: List[str] = None) -> web.Application:
return setup_openapi(
web.Application(),
Path(__file__).parent / "openapi.yaml",
operations,
)
If your OpenAPI schema contains multiple servers schemas, like,
.. code-block:: yaml
servers:
- url: "/api/"
description: "Test environment"
- url: "http://localhost:8080/api/"
description: "Dev environment"
- url: "http://prod.url/api/"
description: "Prod environment"
you have 2 options of telling *rororo* how to use specific server URL.
First, is passing ``server_url``, while setting up OpenAPI, for example,
.. code-block:: python
setup_openapi(
web.Application(),
Path(__file__).parent / "openapi.yaml",
operations,
server_url=URL("http://prod.url/api/"),
)
Second, is more complicated as you need to wrap ``aiohttp.web`` application
into :func:`rororo.settings.setup_settings` and mark each server with
``x-rororo-level`` special key in server schema definition as,
.. code-block:: yaml
servers:
- url: "/api/"
x-rororo-level: "test"
- url: "http://localhost:8080/api/"
x-rororo-level: "dev"
- url: "http://prod.url/api/"
x-rororo-level: "prod"
After, *rororo* will try to equal current app settings level with the
schema and if URL matched, will use given server URL for finding out
route prefix.
By default, *rororo* will validate operation responses against OpenAPI
schema. To disable this feature, pass ``is_validate_response`` falsy flag.
By default, *rororo* will share the OpenAPI schema which is registered
for your aiohttp.web application. In case if you don't want to share this
schema, pass ``has_openapi_schema_handler=False`` on setting up OpenAPI.
By default, *rororo* will enable
:func:`aiohttp_middlewares.cors.cors_middleware` without any settings and
:func:`aiohttp_middlewares.error.error_middleware` with custom error
handler to ensure that security / validation errors does not provide any
mess to stdout. Pass ``use_cors_middleware`` /
``use_error_middleware`` to change or entirely disable this default
behaviour.
For passing custom options to CORS middleware, use
``cors_middleware_kwargs`` mapping. If kwarg does not support by CORS
middleware - *rororo* will raise a ``ConfigurationError``. All list of
options available at documentation for
:func:`aiohttp_middlewares.cors.cors_middleware`.
To simplify things *rororo* expects on OpenAPI 3 path and do reading schema
from file and specifying ``openapi_core.schema.specs.models.Spec`` instance
inside of :func:`rororo.openapi.setup_openapi` call.
However, it is possible to completely customize this default behaviour and
pass OpenAPI ``schema`` and ``spec`` instance directly. In that case
``schema`` keyword argument should contains raw OpenAPI 3 schema as
``Dict[str, Any]``, while ``spec`` to be an
``openapi_core.schema.specs.models.Spec`` instance.
This behaviour might be helpful if you'd like to cache reading schema and
instantiating spec within tests or other environments, which requires
multiple :func:`rororo.openapi.setup_openapi` calls.
.. code-block:: python
from pathlib import Path
import yaml
from aiohttp import web
from openapi_core.shortcuts import create_spec
from rororo import setup_openapi
# Reusable OpenAPI data
openapi_yaml = Path(__file__).parent / "openapi.yaml"
schema = yaml.load(
openapi_yaml.read_bytes(), Loader=yaml.CSafeLoader
)
spec = create_spec(schema)
# Create OpenAPI 3 aiohttp.web server application
app = setup_openapi(web.Application(), schema=schema, spec=spec)
For default behaviour, with passing ``schema_path``, there are few options
on customizing schema load process as well,
By default, *rororo* will use :func:`json.loads` to load OpenAPI schema
content from JSON file and ``yaml.CSafeLoader`` if it is available to load
schema content from YAML files (with fallback to ``yaml.SafeLoader``). But,
for performance considreations, you might use any other function to load
the schema. Example below illustrates how to use ``ujson.loads`` function
to load content from JSON schema,
.. code-block:: python
import ujson
app = setup_openapi(
web.Application(),
Path(__file__).parent / "openapi.json",
operations,
schema_loader=ujson.loads,
)
Schema loader function expects ``bytes`` as only argument and should return
``Dict[str, Any]`` as OpenAPI schema dict.
.. danger::
By default *rororo* does not cache slow calls to read OpenAPI schema
and creating its spec. But sometimes, for example in tests, it is
sufficient to cache those calls. To enable cache behaviour pass
``cache_create_schema_and_spec=True`` or even better,
``cache_create_schema_and_spec=settings.is_test``.
But this may result in unexpected issues, as schema and spec will be
cached once and on next call it will result cached data instead to
attempt read fresh schema from the disk and instantiate OpenAPI Spec
instance.
By default, *rororo* using ``validate_email`` function from
`email-validator <https://github.com/JoshData/python-email-validator>`_
library to validate email strings, which has been declared in OpenAPI
schema as,
.. code-block:: yaml
components:
schemas:
Email:
type: "string"
format: "email"
In most cases ``validate_email(email)`` call should be enough, but in case
if you need to pass extra ``**kwargs`` for validating email strings, setup
``validate_email_kwargs`` such as,
.. code-block:: python
app = setup_openapi(
web.Application(),
Path(__file__).parent / "openapi.json",
operations,
validate_email_kwargs={"check_deliverability": False},
)
"""
if isinstance(schema_path, OperationTableDef):
operations = (schema_path, *operations)
schema_path = None
if schema is None and spec is None:
if schema_path is None:
raise ConfigurationError(
"Please supply only `spec` keyword argument, or only "
"`schema_path` positional argument, not both."
)
# Ensure OpenAPI schema is a readable file
path = (
Path(schema_path) if isinstance(schema_path, str) else schema_path
)
if not path.exists() or not path.is_file():
uid = os.getuid()
raise ConfigurationError(
f"Unable to find OpenAPI schema file at {path}. Please check "
"that file exists at given path and readable by current user "
f"ID: {uid}"
)
# Create the spec and put it to the application dict as well
create_func: CreateSchemaAndSpec = (
create_schema_and_spec_with_cache # type: ignore[assignment]
if cache_create_schema_and_spec
else create_schema_and_spec
)
try:
schema, spec = create_func(path, schema_loader=schema_loader)
except Exception:
raise ConfigurationError(
f"Unable to load valid OpenAPI schema in {path}. In most "
"cases it means that given file doesn't contain valid OpenAPI "
"3 schema. To get full details about errors run "
f"`openapi-spec-validator {path.absolute()}`"
)
elif schema_path is not None:
warnings.warn(
"You supplied `schema_path` positional argument as well as "
"supplying `schema` & `spec` keyword arguments. `schema_path` "
"will be ignored in favor of `schema` & `spec` args.",
stacklevel=2,
)
# Fix all operation securities within OpenAPI spec
spec = fix_spec_operations(spec, cast(DictStrAny, schema))
# Store schema, spec, and validate email kwargs in application dict
app[APP_OPENAPI_SCHEMA_KEY] = schema
app[APP_OPENAPI_SPEC_KEY] = spec
app[APP_VALIDATE_EMAIL_KWARGS_KEY] = validate_email_kwargs
# Register the route to dump openapi schema used for the application if
# required
route_prefix = find_route_prefix(
cast(DictStrAny, schema),
server_url=server_url,
settings=app.get(APP_SETTINGS_KEY),
)
if has_openapi_schema_handler:
app.router.add_get(
add_prefix("/openapi.{schema_format}", route_prefix),
views.openapi_schema,
)
# Register all operation handlers to web application
for item in operations:
app.router.add_routes(
convert_operations_to_routes(item, spec, prefix=route_prefix)
)
# Add OpenAPI middleware
kwargs = error_middleware_kwargs or {}
kwargs.setdefault("default_handler", views.default_error_handler)
try:
app.middlewares.insert(
0,
openapi_middleware(
is_validate_response=is_validate_response,
use_error_middleware=use_error_middleware,
error_middleware_kwargs=kwargs,
),
)
except TypeError:
raise ConfigurationError(
"Unsupported kwargs passed to error middleware. Please check "
"given kwargs and remove unsupported ones: "
f"{error_middleware_kwargs!r}"
)
# Add CORS middleware if necessary
if use_cors_middleware:
try:
app.middlewares.insert(
0, cors_middleware(**(cors_middleware_kwargs or {}))
)
except TypeError:
raise ConfigurationError(
"Unsupported kwargs passed to CORS middleware. Please check "
"given kwargs and remove unsupported ones: "
f"{cors_middleware_kwargs!r}"
)
return app