mypy - use Optional & Union for python <3.10 compatibility

aiopenapi3/_types.py

@@ -1,15 +1,6 @@
 from . import v20, v30, v31
 
-from typing import (
-    TYPE_CHECKING,
-    Dict,
-    List,
-    Sequence,
-    Tuple,
-    Union,
-    TypeAlias,
-    Type,
-)
+from typing import TYPE_CHECKING, Dict, List, Sequence, Tuple, Union, TypeAlias, Type, Optional
 
 import yaml
 
@@ -23,7 +14,7 @@
 RequestFileParameter = Tuple[str, FileTypes]
 RequestFilesParameter = Sequence[RequestFileParameter]
 
-JSON: TypeAlias = dict[str, "JSON"] | list["JSON"] | str | int | float | bool | None
+JSON: TypeAlias = Optional[Union[dict[str, "JSON"], list["JSON"], str, int, float, bool]]
 """
 Define a JSON type
 https://github.com/python/typing/issues/182#issuecomment-1320974824

aiopenapi3/base.py

@@ -394,8 +394,8 @@ def _get_identity(self, prefix="XLS", name=None):
 
     def set_type(
         self,
-        names: List[str] | None = None,
-        discriminators: Sequence[DiscriminatorBase] | None = None,
+        names: Optional[List[str]] = None,
+        discriminators: Optional[Sequence[DiscriminatorBase]] = None,
         extra: Optional["SchemaBase"] = None,
     ) -> Type[BaseModel]:
         from .model import Model
@@ -417,8 +417,8 @@ def set_type(
 
     def get_type(
         self,
-        names: List[str] | None = None,
-        discriminators: Sequence[DiscriminatorBase] | None = None,
+        names: Optional[List[str]] = None,
+        discriminators: Optional[Sequence[DiscriminatorBase]] = None,
         extra: Optional["SchemaBase"] = None,
         fwdref: bool = False,
     ) -> Union[Type[BaseModel], ForwardRef]:
@@ -462,7 +462,7 @@ def model(self, data: "JSON") -> Union[BaseModel, List[BaseModel]]:
 
 
 class OperationBase:
-    # parameters: Optional[List[ParameterBase | ReferenceBase]]
+    # parameters: Optional[List[Union[ParameterBase, ReferenceBase]]]
     parameters: List[Any]
 
     def _validate_path_parameters(self, pi_: "PathItemBase", path_: str, loc: Tuple[Any, str]):

aiopenapi3/cli.py

@@ -12,7 +12,6 @@
 import tracemalloc
 import linecache
 import logging
-from typing import Callable
 
 import jmespath
 import yaml
@@ -34,8 +33,11 @@
 from .loader import ChainLoader, RedirectLoader, WebLoader
 import aiopenapi3.loader
 
+from .log import init
 
-log: Callable[[...], None] | None = None
+init()
+
+# log: Callable[..., None] | None = None
 
 
 def loader_prepare(args, session_factory):
@@ -62,7 +64,7 @@ def plugins_load(baseurl, plugins: List[str]) -> List[aiopenapi3.plugin.Plugin]:
             raise ValueError("importlib")
         if (module := importlib.util.module_from_spec(spec)) is None:
             raise ValueError("importlib")
-        assert spec and module
+        assert spec and spec.loader and module
         spec.loader.exec_module(module)
         for c in clsp:
             plugin = getattr(module, c)

aiopenapi3/loader.py

@@ -2,7 +2,7 @@
 import json
 import logging
 import typing
-
+from typing import Optional
 import yaml
 import httpx
 import yarl
@@ -112,7 +112,7 @@ def __init__(self, yload: "YAMLLoaderType" = YAML12Loader):
         self.yload = yload
 
     @abc.abstractmethod
-    def load(self, plugins: Plugins, url: yarl.URL, codec: str | None = None):
+    def load(self, plugins: Plugins, url: yarl.URL, codec: Optional[str] = None):
         """
         load and decode description document
 
@@ -124,7 +124,7 @@ def load(self, plugins: Plugins, url: yarl.URL, codec: str | None = None):
         raise NotImplementedError("load")
 
     @classmethod
-    def decode(cls, data: bytes, codec: str | None) -> str:
+    def decode(cls, data: bytes, codec: Optional[str]) -> str:
         """
         decode bytes to ascii or utf-8
 
@@ -196,7 +196,7 @@ class NullLoader(Loader):
     Loader does not load anything
     """
 
-    def load(self, plugins: Plugins, url: yarl.URL, codec: str | None = None):
+    def load(self, plugins: Plugins, url: yarl.URL, codec: Optional[str] = None):
         raise NotImplementedError("load")
 
 
@@ -211,7 +211,7 @@ def __init__(self, baseurl: yarl.URL, session_factory=httpx.Client, yload: "YAML
         self.baseurl: yarl.URL = baseurl
         self.session_factory = session_factory
 
-    def load(self, plugins: Plugins, url: yarl.URL, codec: str | None = None) -> "JSON":
+    def load(self, plugins: Plugins, url: yarl.URL, codec: Optional[str] = None) -> "JSON":
         url = self.baseurl.join(url)
         with self.session_factory() as session:
             data = session.get(str(url))
@@ -239,7 +239,7 @@ def __init__(self, base: Path, yload: "YAMLLoaderType" = YAML12Loader):
         assert isinstance(base, Path)
         self.base = base
 
-    def load(self, plugins: Plugins, url: yarl.URL, codec: str | None = None):
+    def load(self, plugins: Plugins, url: yarl.URL, codec: Optional[str] = None):
         assert isinstance(url, yarl.URL)
         assert plugins
         file = Path(url.path)
@@ -262,7 +262,7 @@ class RedirectLoader(FileSystemLoader):
     everything but the "name" is stripped of the url
     """
 
-    def load(self, plugins: "Plugins", url: yarl.URL, codec: str | None = None):
+    def load(self, plugins: "Plugins", url: yarl.URL, codec: Optional[str] = None):
         return super().load(plugins, yarl.URL(url.name), codec)
 
 
@@ -280,7 +280,7 @@ def __init__(self, *loaders, yload: "YAMLLoaderType" = YAML12Loader):
         Loader.__init__(self, yload)
         self.loaders = loaders
 
-    def load(self, plugins: "Plugins", url: yarl.URL, codec: str | None = None):
+    def load(self, plugins: "Plugins", url: yarl.URL, codec: Optional[str] = None):
         log.debug(f"load {url}")
         errors = []
         for i in self.loaders:

aiopenapi3/log.py

@@ -1,14 +1,14 @@
 import sys
 import logging.config
 import os
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional
 
 if sys.version_info >= (3, 9):
     from pathlib import Path
 else:
     from pathlib3x import Path
 
-handlers: List[str] | None = None
+handlers: Optional[List[str]] = None
 
 
 def init(force: bool = False) -> None:

aiopenapi3/model.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 import collections
 import dataclasses
 import logging
@@ -114,9 +112,9 @@ class Model:  # (BaseModel):
     def from_schema(
         cls,
         schema: "SchemaType",
-        schemanames: List[str] | None = None,
-        discriminators: List["DiscriminatorType"] | None = None,
-        extra: "SchemaType" | None = None,
+        schemanames: Optional[List[str]] = None,
+        discriminators: Optional[List["DiscriminatorType"]] = None,
+        extra: Optional["SchemaType"] = None,
     ) -> Type[BaseModel]:
         if schemanames is None:
             schemanames = []
@@ -322,7 +320,7 @@ def configof(schema: "SchemaType"):
 
     @staticmethod
     def typeof(
-        schema: Optional[Union["SchemaType", "ReferenceType"]], _type: str | None = None, fwdref: bool = False
+        schema: Optional[Union["SchemaType", "ReferenceType"]], _type: Optional[str] = None, fwdref: bool = False
     ) -> Type:
         if schema is None:
             return BaseModel
@@ -517,7 +515,7 @@ def is_type(schema: "SchemaType", type_) -> bool:
         return isinstance(schema.type, str) and schema.type == type_ or Model.or_type(schema, type_, l=None)
 
     @staticmethod
-    def or_type(schema: "SchemaType", type_: str, l: int | None = 2) -> bool:
+    def or_type(schema: "SchemaType", type_: str, l: Optional[int] = 2) -> bool:
         return isinstance((t := schema.type), list) and (l is None or len(t) == l) and type_ in t
 
     @staticmethod

aiopenapi3/openapi.py

@@ -87,8 +87,8 @@ def load_sync(
         cls,
         url,
         session_factory: Callable[..., httpx.Client] = httpx.Client,
-        loader: Loader | None = None,
-        plugins: List[Plugin] | None = None,
+        loader: Optional[Loader] = None,
+        plugins: Optional[List[Plugin]] = None,
         use_operation_tags: bool = False,
     ) -> "OpenAPI":
         """
@@ -110,8 +110,8 @@ async def load_async(
         cls,
         url: str,
         session_factory: Callable[..., httpx.AsyncClient] = httpx.AsyncClient,
-        loader: Loader | None = None,
-        plugins: List[Plugin] | None = None,
+        loader: Optional[Loader] = None,
+        plugins: Optional[List[Plugin]] = None,
         use_operation_tags: bool = False,
     ) -> "OpenAPI":
         """
@@ -139,8 +139,8 @@ def load_file(
         url: str,
         path: Union[str, pathlib.Path, yarl.URL],
         session_factory: Callable[..., Union[httpx.AsyncClient, httpx.Client]] = httpx.AsyncClient,
-        loader: Loader | None = None,
-        plugins: List[Plugin] | None = None,
+        loader: Optional[Loader] = None,
+        plugins: Optional[List[Plugin]] = None,
         use_operation_tags: bool = False,
     ) -> "OpenAPI":
         """
@@ -166,8 +166,8 @@ def loads(
         url: str,
         data: str,
         session_factory: Callable[..., Union[httpx.AsyncClient, httpx.Client]] = httpx.AsyncClient,
-        loader: Loader | None = None,
-        plugins: List[Plugin] | None = None,
+        loader: Optional[Loader] = None,
+        plugins: Optional[List[Plugin]] = None,
         use_operation_tags: bool = False,
     ) -> "OpenAPI":
         """
@@ -213,8 +213,8 @@ def __init__(
         url: str,
         document: "JSON",
         session_factory: Callable[..., Union[httpx.Client, httpx.AsyncClient]] = httpx.AsyncClient,
-        loader: Loader | None = None,
-        plugins: List[Plugin] | None = None,
+        loader: Optional[Loader] = None,
+        plugins: Optional[List[Plugin]] = None,
         use_operation_tags: bool = True,
     ) -> None:
         """
@@ -233,7 +233,7 @@ def __init__(
 
         self._session_factory: Callable[..., Union[httpx.Client, httpx.AsyncClient]] = session_factory
 
-        self.loader: Loader | None = loader
+        self.loader: Optional[Loader] = loader
         """
         Loader - loading referenced documents
         """
@@ -722,7 +722,7 @@ def __copy__(self) -> "OpenAPI":
         api.loader = self.loader
         return api
 
-    def clone(self, baseurl: yarl.URL | None = None) -> "OpenAPI":
+    def clone(self, baseurl: Optional[yarl.URL] = None) -> "OpenAPI":
         """
         shallwo copy the api object
         optional set a base url
@@ -735,7 +735,7 @@ def clone(self, baseurl: yarl.URL | None = None) -> "OpenAPI":
         return api
 
     @staticmethod
-    def cache_load(path: pathlib.Path, plugins: List[Plugin] | None = None, session_factory=None) -> "OpenAPI":
+    def cache_load(path: pathlib.Path, plugins: Optional[List[Plugin]] = None, session_factory=None) -> "OpenAPI":
         """
         read a pickle api object from path and init the schema types
 

aiopenapi3/plugin.py

@@ -49,9 +49,9 @@ class Init(Plugin):
     class Context:
         initialized: Optional["OpenAPI"] = None
         """available in :func:`~aiopenapi3.plugin.Init.initialized`"""
-        schemas: Dict[str, "SchemaBase"] | None = None
+        schemas: Optional[Dict[str, "SchemaBase"]] = None
         """available in :func:`~aiopenapi3.plugin.Init.schemas`"""
-        paths: Dict[str, "PathItemBase"] | None = None
+        paths: Optional[Dict[str, "PathItemBase"]] = None
         """available in :func:`~aiopenapi3.plugin.Init.paths`"""
 
     def schemas(self, ctx: "Init.Context") -> "Init.Context":  # pragma: no cover
@@ -197,7 +197,7 @@ def __init__(self, plugins: List[Plugin]):
         self._message = self._get_domain("message", plugins)
 
     def _get_domain(self, name: str, plugins: List[Plugin]) -> "Domain":
-        domain: Type[Plugin] | None
+        domain: Optional[Type[Plugin]]
         if (domain := self._domains.get(name)) is None:
             raise ValueError(name)  # noqa
 

aiopenapi3/request.py

@@ -10,6 +10,7 @@
 
 from aiopenapi3.errors import ContentLengthExceededError
 
+
 try:
     from contextlib import aclosing
 except:  # <= Python 3.10
@@ -48,7 +49,7 @@ async def aclosing(thing):
 
 
 class RequestParameter:
-    def __init__(self, url: yarl.URL | str):
+    def __init__(self, url: Union[yarl.URL, str]):
         self.url: str = str(url)
         self.auth: Optional["AuthTypes"] = None
         self.cookies: Dict[str, str] = {}

aiopenapi3/v20/glue.py

@@ -283,7 +283,9 @@ def _process_stream(self, result: httpx.Response) -> Tuple[Dict[str, str], Optio
         headers = self._process__headers(result, result.headers, expected_response)
         return headers, expected_response.schema_
 
-    def _process_request(self, result: httpx.Response) -> Tuple[Dict[str, str], pydantic.BaseModel | str | None]:
+    def _process_request(
+        self, result: httpx.Response
+    ) -> Tuple[Dict[str, str], Optional[Union[pydantic.BaseModel, str]]]:
         rheaders = dict()
         # spec enforces these are strings
         status_code = str(result.status_code)

aiopenapi3/v30/glue.py

@@ -463,7 +463,7 @@ def _process__headers(
         return rheaders
 
     def _process__content_type(
-        self, result: httpx.Response, expected_response: "v3xResponseType", content_type: str | None
+        self, result: httpx.Response, expected_response: "v3xResponseType", content_type: Optional[str]
     ) -> Tuple[str, "v3xMediaTypeType"]:
         if content_type:
             content_type, _, encoding = content_type.partition(";")
@@ -501,7 +501,9 @@ def _process_stream(self, result: httpx.Response) -> Tuple[Dict[str, str], Optio
 
         return headers, expected_media.schema_
 
-    def _process_request(self, result: httpx.Response) -> Tuple[Dict[str, str], pydantic.BaseModel | str | None]:
+    def _process_request(
+        self, result: httpx.Response
+    ) -> Tuple[Dict[str, str], Optional[Union[pydantic.BaseModel, str]]]:
         rheaders = dict()
         # spec enforces these are strings
         status_code = str(result.status_code)

aiopenapi3/v30/parameter.py

@@ -275,9 +275,9 @@ def validate_Parameter(cls, p: "ParameterBase"):
 def encode_parameter(
     name: str,
     value: object,
-    style: str | None,
-    explode: bool | None,
-    allowReserved: bool | None,
+    style: Optional[str],
+    explode: Optional[bool],
+    allowReserved: Optional[bool],
     in_: str,
     schema_: Schema,
 ) -> Union[str, bytes]:

aiopenapi3/v30/servers.py

@@ -13,7 +13,7 @@ class ServerVariable(ObjectExtended):
     """
 
     enum: Optional[List[str]] = Field(default=None)
-    default: str | None = Field(...)
+    default: Optional[str] = Field(...)
     description: Optional[str] = Field(default=None)
 
 

aiopenapi3/v31/servers.py

@@ -13,7 +13,7 @@ class ServerVariable(ObjectExtended):
     """
 
     enum: Optional[List[str]] = Field(default=None)
-    default: str | None = Field(...)
+    default: Optional[str] = Field(...)
     description: Optional[str] = Field(default=None)
 
     @model_validator(mode="after")