feat(openapi): Allow to pass kwargs to validate email func (#135)

*rororo* using [email-validator](https://github.com/JoshData/python-email-validator)
library for validating email strings.

In most cases `validate_email(email)` call should be enough, but when
not it is now possible to pass `validate_email_kwargs` dict such as
follows,

```python
app = setup_openapi(
    web.Application(),
    Path(__file__).parent / "openapi.json",
    operations,
    validate_email_kwargs={"check_deliverability": False},
)
```

Fixes: #133

src/rororo/openapi/annotations.py

@@ -1,4 +1,14 @@
-from typing import Dict, List
+from typing import Any, Dict, List, Union
+
+from ..annotations import TypedDict
 
 
 SecurityDict = Dict[str, List[str]]
+
+
+class ValidateEmailKwargsDict(TypedDict, total=False):
+    allow_smtputf8: bool
+    allow_empty_local: bool
+    check_deliverability: bool
+    timeout: Union[int, float]
+    dns_resolver: Any

src/rororo/openapi/constants.py

@@ -4,6 +4,9 @@
 #: Key to store OpenAPI spec within the ``web.Application`` instance
 APP_OPENAPI_SPEC_KEY = "rororo_openapi_spec"
 
+#: Key to store kwargs to pass to ``validate_email`` function
+APP_VALIDATE_EMAIL_KWARGS_KEY = "rororo_validate_email_kwargs"
+
 #: Key to store request method -> operation ID mapping in handler
 HANDLER_OPENAPI_MAPPING_KEY = "__rororo_openapi_mapping__"
 

src/rororo/openapi/core_validators.py

@@ -49,6 +49,7 @@
 )
 from openapi_schema_validator._format import oas30_format_checker
 
+from .annotations import ValidateEmailKwargsDict
 from .data import OpenAPIParameters, to_openapi_parameters
 from .exceptions import CastError, ValidationError
 from .security import validate_security
@@ -83,9 +84,14 @@ class EmailFormatter(Formatter):
     to ensure that given string is a valid email.
     """
 
+    kwargs: ValidateEmailKwargsDict
+
+    def __init__(self, kwargs: ValidateEmailKwargsDict = None) -> None:
+        self.kwargs: ValidateEmailKwargsDict = kwargs or {}
+
     def validate(self, value: str) -> bool:
         try:
-            validate_email(value)
+            validate_email(value, **self.kwargs)
         except EmailNotValidError as err:
             raise FormatError(f"{value!r} is not an 'email'", cause=err)
         return True
@@ -167,9 +173,6 @@ def get_formatter(
         return super().get_formatter(default_formatters, type_format)
 
 
-CUSTOM_FORMATTERS = {"email": EmailFormatter()}
-
-
 class BaseValidator(CoreBaseValidator):
     """Custom base validator to deal with tz aware date time strings.
 
@@ -261,16 +264,29 @@ def _unmarshal(  # type: ignore
         )
 
 
+def get_custom_formatters(
+    *, validate_email_kwargs: ValidateEmailKwargsDict = None
+) -> Dict[str, Formatter]:
+    return {"email": EmailFormatter(validate_email_kwargs)}
+
+
 def validate_core_request(
-    spec: Spec, core_request: OpenAPIRequest
+    spec: Spec,
+    core_request: OpenAPIRequest,
+    *,
+    validate_email_kwargs: ValidateEmailKwargsDict = None,
 ) -> Tuple[MappingStrAny, OpenAPIParameters, Any]:
     """
     Instead of validating request parameters & body in two calls, validate them
     at once with passing custom formatters.
     """
+    custom_formatters = get_custom_formatters(
+        validate_email_kwargs=validate_email_kwargs
+    )
+
     validator = RequestValidator(
         spec,
-        custom_formatters=CUSTOM_FORMATTERS,
+        custom_formatters=custom_formatters,
         base_url=get_base_url(core_request),
     )
     result = validator.validate(core_request)
@@ -289,11 +305,17 @@ def validate_core_response(
     spec: Spec,
     core_request: OpenAPIRequest,
     core_response: OpenAPIResponse,
+    *,
+    validate_email_kwargs: ValidateEmailKwargsDict = None,
 ) -> Any:
     """Pass custom formatters for validating response data."""
+    custom_formatters = get_custom_formatters(
+        validate_email_kwargs=validate_email_kwargs
+    )
+
     validator = ResponseValidator(
         spec,
-        custom_formatters=CUSTOM_FORMATTERS,
+        custom_formatters=custom_formatters,
         base_url=get_base_url(core_request),
     )
     result = validator.validate(core_request, core_response)

src/rororo/openapi/openapi.py

@@ -20,16 +20,23 @@
 import yaml
 from aiohttp import hdrs, web
 from aiohttp_middlewares import cors_middleware
+from aiohttp_middlewares.annotations import (
+    ExceptionType,
+    StrCollection,
+    UrlCollection,
+)
+from aiohttp_middlewares.error import Config as ErrorMiddlewareConfig
 from openapi_core.schema.specs.models import Spec
 from openapi_core.shortcuts import create_spec
 from pyrsistent import pmap
 from yarl import URL
 
 from . import views
-from .annotations import SecurityDict
+from .annotations import SecurityDict, ValidateEmailKwargsDict
 from .constants import (
     APP_OPENAPI_SCHEMA_KEY,
     APP_OPENAPI_SPEC_KEY,
+    APP_VALIDATE_EMAIL_KWARGS_KEY,
     HANDLER_OPENAPI_MAPPING_KEY,
 )
 from .core_data import get_core_operation
@@ -42,6 +49,7 @@
     F,
     Handler,
     Protocol,
+    TypedDict,
     ViewType,
 )
 from ..settings import APP_SETTINGS_KEY, BaseSettings
@@ -51,13 +59,32 @@
 Url = Union[str, URL]
 
 
+class CorsMiddlewareKwargsDict(TypedDict, total=False):
+    allow_all: bool
+    origins: Optional[UrlCollection]
+    urls: Optional[UrlCollection]
+    expose_headers: Optional[UrlCollection]
+    allow_headers: StrCollection
+    allow_methods: StrCollection
+    allow_credentials: bool
+    max_age: Optional[int]
+
+
 class CreateSchemaAndSpec(Protocol):
     def __call__(
         self, path: Path, *, schema_loader: SchemaLoader = None
     ) -> Tuple[DictStrAny, Spec]:  # pragma: no cover
         ...
 
 
+class ErrorMiddlewareKwargsDict(TypedDict, total=False):
+    default_handler: Handler
+    config: Optional[ErrorMiddlewareConfig]
+    ignore_exceptions: Optional[
+        Union[ExceptionType, Tuple[ExceptionType, ...]]
+    ]
+
+
 @attr.dataclass(slots=True)
 class OperationTableDef:
     """Map OpenAPI 3 operations to aiohttp.web view handlers.
@@ -409,11 +436,12 @@ def setup_openapi(
     is_validate_response: bool = True,
     has_openapi_schema_handler: bool = True,
     use_error_middleware: bool = True,
-    error_middleware_kwargs: DictStrAny = None,
+    error_middleware_kwargs: ErrorMiddlewareKwargsDict = None,
     use_cors_middleware: bool = True,
-    cors_middleware_kwargs: DictStrAny = None,
+    cors_middleware_kwargs: CorsMiddlewareKwargsDict = None,
     schema_loader: SchemaLoader = None,
     cache_create_schema_and_spec: bool = False,
+    validate_email_kwargs: ValidateEmailKwargsDict = None,
 ) -> web.Application:  # pragma: no cover
     ...
 
@@ -428,9 +456,10 @@ def setup_openapi(
     is_validate_response: bool = True,
     has_openapi_schema_handler: bool = True,
     use_error_middleware: bool = True,
-    error_middleware_kwargs: DictStrAny = None,
+    error_middleware_kwargs: ErrorMiddlewareKwargsDict = None,
     use_cors_middleware: bool = True,
-    cors_middleware_kwargs: DictStrAny = None,
+    cors_middleware_kwargs: CorsMiddlewareKwargsDict = None,
+    validate_email_kwargs: ValidateEmailKwargsDict = None,
 ) -> web.Application:  # pragma: no cover
     ...
 
@@ -450,6 +479,7 @@ def setup_openapi(  # type: ignore
     cors_middleware_kwargs: DictStrAny = None,
     schema_loader: SchemaLoader = None,
     cache_create_schema_and_spec: bool = False,
+    validate_email_kwargs: ValidateEmailKwargsDict = None,
 ) -> web.Application:
     """Setup OpenAPI schema to use with aiohttp.web application.
 
@@ -626,6 +656,33 @@ def create_app(argv: List[str] = None) -> web.Application:
         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):
@@ -677,9 +734,10 @@ def create_app(argv: List[str] = None) -> web.Application:
     # Fix all operation securities within OpenAPI spec
     spec = fix_spec_operations(spec, cast(DictStrAny, schema))
 
-    # Store schema and spec in application dict
+    # 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

src/rororo/openapi/utils.py

@@ -6,9 +6,11 @@
 from openapi_core.validation.request.datatypes import OpenAPIRequest
 from yarl import URL
 
+from .annotations import ValidateEmailKwargsDict
 from .constants import (
     APP_OPENAPI_SCHEMA_KEY,
     APP_OPENAPI_SPEC_KEY,
+    APP_VALIDATE_EMAIL_KWARGS_KEY,
     REQUEST_OPENAPI_CONTEXT_KEY,
 )
 from .data import OpenAPIContext, OpenAPIParameters
@@ -81,6 +83,20 @@ def get_openapi_spec(mixed: Union[web.Application, ChainMapProxy]) -> Spec:
         )
 
 
+def get_validate_email_kwargs(
+    mixed: Union[web.Application, ChainMapProxy]
+) -> ValidateEmailKwargsDict:
+    try:
+        return cast(
+            ValidateEmailKwargsDict, mixed[APP_VALIDATE_EMAIL_KWARGS_KEY] or {}
+        )
+    except KeyError:
+        raise ConfigurationError(
+            "Seems like kwargs to pass to validate_email function is not "
+            "registered for the given application"
+        )
+
+
 def get_validated_data(request: web.Request) -> Any:
     """Shortcut to get validated data (request body) for given request.
 

src/rororo/openapi/validators.py

@@ -4,7 +4,7 @@
 from .core_data import to_core_openapi_request, to_core_openapi_response
 from .core_validators import validate_core_request, validate_core_response
 from .data import OpenAPIContext
-from .utils import get_openapi_spec
+from .utils import get_openapi_spec, get_validate_email_kwargs
 
 
 async def validate_request(request: web.Request) -> web.Request:
@@ -14,7 +14,9 @@ async def validate_request(request: web.Request) -> web.Request:
     request[REQUEST_CORE_REQUEST_KEY] = core_request
 
     security, parameters, data = validate_core_request(
-        get_openapi_spec(config_dict), core_request
+        get_openapi_spec(config_dict),
+        core_request,
+        validate_email_kwargs=get_validate_email_kwargs(config_dict),
     )
     request[REQUEST_OPENAPI_CONTEXT_KEY] = OpenAPIContext(
         request=request,
@@ -31,9 +33,12 @@ async def validate_request(request: web.Request) -> web.Request:
 def validate_response(
     request: web.Request, response: web.StreamResponse
 ) -> web.StreamResponse:
+    config_dict = request.config_dict
+
     validate_core_response(
-        get_openapi_spec(request.config_dict),
+        get_openapi_spec(config_dict),
         request[REQUEST_CORE_REQUEST_KEY],
         to_core_openapi_response(response),
+        validate_email_kwargs=get_validate_email_kwargs(config_dict),
     )
     return response

tests/rororo/test_openapi_core_validators.py

@@ -17,3 +17,12 @@ def test_invalid_value(invalid_value):
 @pytest.mark.parametrize("value", ("email@domain.com", "EmAiL@domain.com"))
 def test_valid_email(value):
     assert EmailFormatter().validate(value) is True
+
+
+def test_valid_email_without_check_deliverability():
+    assert (
+        EmailFormatter(kwargs={"check_deliverability": False}).validate(
+            "email@sub.domain.com"
+        )
+        is True
+    )