style: Introduce no-optional pre-commit hook (#295)

To replace `Optional[T]` type annotations in code base to `Union[T, None]`.

Author: playpauseandstop

.pre-commit-config.yaml

@@ -12,6 +12,12 @@ repos:
         language_version: *python_version
         stages: ["commit-msg"]
 
+  - repo: "https://github.com/Kludex/no-optional"
+    rev: "0.4.0"
+    hooks:
+      - id: "no_optional"
+        name: "Format code (no-optional)"
+
   - repo: "https://github.com/PyCQA/isort"
     rev: "5.10.1"
     hooks:
@@ -84,7 +90,7 @@ repos:
   - repo: "https://github.com/asottile/yesqa"
     rev: "v1.4.0"
     hooks:
-      - id: yesqa
+      - id: "yesqa"
         name: "Lint code (yesqa)"
         additional_dependencies: *flake8_additional_dependencies
         exclude: ^docs/.*$

examples/petstore/src/petstore/data.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Union
 
 import attr
 
@@ -8,7 +8,7 @@
 @attr.dataclass(frozen=True, slots=True)
 class NewPet:
     name: str
-    tag: Optional[str]
+    tag: Union[str, None]
 
     def to_pet(self, pet_id: int) -> "Pet":
         return Pet(id=pet_id, name=self.name, tag=self.tag)

src/rororo/aio.py

@@ -9,7 +9,7 @@
 """
 
 from contextlib import contextmanager
-from typing import Iterator, Optional, Union
+from typing import Iterator, Union
 from urllib.parse import urlparse
 
 from aiohttp import web
@@ -121,7 +121,7 @@ def is_xhr_request(request: web.Request) -> bool:
 
     :param request: Request instance.
     """
-    value: Optional[str] = request.headers.get("X-Requested-With")
+    value: Union[str, None] = request.headers.get("X-Requested-With")
     return value == "XMLHttpRequest"
 
 
@@ -140,7 +140,7 @@ async def connect_redis(url=None):
     """
     parts = urlparse(url)
 
-    db: Optional[Union[str, int]] = parts.path[1:] or None
+    db: Union[str, int, None] = parts.path[1:] or None
     if db:
         db = int(db)
 

src/rororo/logger.py

@@ -9,7 +9,7 @@
 
 import logging
 import sys
-from typing import Any, Optional, Union
+from typing import Any, Union
 
 from rororo.annotations import DictStrAny
 
@@ -81,7 +81,7 @@ def default_logging_dict(*loggers: str, **kwargs: Any) -> DictStrAny:
 
 def update_sentry_logging(
     logging_dict: DictStrAny,
-    sentry_dsn: Optional[str],
+    sentry_dsn: Union[str, None],
     *loggers: str,
     level: Union[str, int, None] = None,
     **kwargs: Any,

src/rororo/openapi/core_data.py

@@ -1,4 +1,4 @@
-from typing import cast, Optional, Union
+from typing import cast, Union
 
 from aiohttp import hdrs, web
 from aiohttp.payload import IOBasePayload, Payload
@@ -19,7 +19,7 @@
 
 def find_core_operation(
     request: web.Request, handler: Handler
-) -> Optional[Operation]:
+) -> Union[Operation, None]:
     mapping = getattr(handler, HANDLER_OPENAPI_MAPPING_KEY, None)
     if not mapping:
         return None
@@ -69,7 +69,7 @@ async def to_core_openapi_request(request: web.Request) -> OpenAPIRequest:
     Afterwards opeanpi-core request can be used for validation request data
     against spec.
     """
-    body: Optional[Union[bytes, str]] = None
+    body: Union[bytes, str, None] = None
     if request.body_exists and request.can_read_body:
         raw_body = await request.read()
 
@@ -100,7 +100,7 @@ def to_core_openapi_response(response: web.StreamResponse) -> OpenAPIResponse:
 
 def to_core_openapi_response_data(
     response: web.StreamResponse,
-) -> Optional[bytes]:
+) -> Union[bytes, None]:
     if isinstance(response, web.Response):
         body = response.body
         if not body:

src/rororo/openapi/core_validators.py

@@ -1,6 +1,6 @@
 from collections import deque
 from functools import partial
-from typing import Any, cast, Dict, Iterator, List, Optional, Tuple, Union
+from typing import Any, cast, Dict, Iterator, List, Tuple, Union
 
 import pyrsistent
 from email_validator import EmailNotValidError, validate_email
@@ -71,7 +71,7 @@ class ArrayUnmarshaller(CoreArrayUnmarshaller):
     arrays: https://github.com/p1c2u/openapi-core/issues/251
     """
 
-    def __call__(self, value: Any = NoValue) -> Optional[List[Any]]:
+    def __call__(self, value: Any = NoValue) -> Union[List[Any], None]:
         if value is None and self.schema.nullable:
             return None
         return cast(List[Any], super().__call__(value))
@@ -108,7 +108,7 @@ class ObjectUnmarshaller(CoreObjectUnmarshaller):
     objects: https://github.com/p1c2u/openapi-core/issues/232
     """
 
-    def __call__(self, value: Any = NoValue) -> Optional[Dict[Any, Any]]:
+    def __call__(self, value: Any = NoValue) -> Union[Dict[Any, Any], None]:
         if value is None and self.schema.nullable:
             return None
         return cast(Dict[Any, Any], super().__call__(value))

src/rororo/openapi/exceptions.py

@@ -2,7 +2,7 @@
 import re
 from contextlib import contextmanager
 from contextvars import ContextVar
-from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
+from typing import Any, Dict, Iterator, List, Tuple, Union
 
 import attr
 from jsonschema.exceptions import ValidationError as JsonSchemaValidationError
@@ -385,7 +385,7 @@ def get_parameter_error_details(
     loc: List[PathItem],
     err: OpenAPIParameterError,
 ) -> ValidationErrorItem:
-    parameter_name: Optional[str] = getattr(err, "name", None)
+    parameter_name: Union[str, None] = getattr(err, "name", None)
     if parameter_name is None:
         return get_common_error_details(loc, err)
 

src/rororo/openapi/openapi.py

@@ -4,17 +4,7 @@
 import warnings
 from functools import lru_cache, partial
 from pathlib import Path
-from typing import (
-    Callable,
-    cast,
-    Deque,
-    Dict,
-    List,
-    Optional,
-    overload,
-    Tuple,
-    Union,
-)
+from typing import Callable, cast, Deque, Dict, List, overload, Tuple, Union
 
 import attr
 import yaml
@@ -344,7 +334,7 @@ def fix_spec_operations(spec: Spec, schema: DictStrAny) -> Spec:
     as unsecured. With missed operation security - use global security schema
     if it is defined.
     """
-    mapping: Dict[str, Optional[SecurityDict]] = {}
+    mapping: Dict[str, Union[SecurityDict, None]] = {}
 
     for path_data in schema["paths"].values():
         for maybe_operation_data in path_data.values():

src/rororo/openapi/security.py

@@ -1,4 +1,4 @@
-from typing import cast, List, Optional, Union
+from typing import cast, List, Union
 
 from aiohttp import BasicAuth, hdrs
 from openapi_core.schema.operations.models import Operation
@@ -29,13 +29,15 @@ def basic_auth_factory(value: str) -> BasicAuth:
     return BasicAuth.decode(f"Basic {value}")
 
 
-def get_jwt_security_data(request: OpenAPIRequest) -> Optional[str]:
+def get_jwt_security_data(request: OpenAPIRequest) -> Union[str, None]:
     """Get JWT bearer security data.
 
     At a moment, openapi-core attempts to decode JWT header using same rules
     as for basic auth, which might return unexpected results.
     """
-    header: Optional[str] = request.parameters.header.get(AUTHORIZATION_HEADER)
+    header: Union[str, None] = request.parameters.header.get(
+        AUTHORIZATION_HEADER
+    )
 
     # Header does not exist
     if header is None:
@@ -54,7 +56,7 @@ def get_jwt_security_data(request: OpenAPIRequest) -> Optional[str]:
 
 def get_security_data(
     validator: RequestValidator, request: OpenAPIRequest, scheme_name: str
-) -> Optional[Union[BasicAuth, str]]:
+) -> Union[BasicAuth, str, None]:
     """Get security data from request.
 
     Currently supported getting API Key & HTTP security data. OAuth & OpenID
@@ -83,7 +85,7 @@ def get_security_list(
 
 def get_security_scheme(
     validator: RequestValidator, scheme_name: str
-) -> Optional[SecurityScheme]:
+) -> Union[SecurityScheme, None]:
     return validator.spec.components.security_schemes.get(scheme_name)
 
 

src/rororo/openapi/utils.py

@@ -1,4 +1,4 @@
-from typing import Any, cast, Optional, Union
+from typing import Any, cast, Union
 
 from aiohttp import web
 from aiohttp.helpers import ChainMapProxy
@@ -18,7 +18,7 @@
 from rororo.openapi.exceptions import ConfigurationError, ContextError
 
 
-def add_prefix(path: str, prefix: Optional[str]) -> str:
+def add_prefix(path: str, prefix: Union[str, None]) -> str:
     if prefix:
         if prefix[-1] == "/":
             prefix = prefix[:-1]