🐛 Typing compatibility with Python 3.7

flama/background.py

@@ -9,14 +9,14 @@
 
 from flama import concurrency
 
-if sys.version_info >= (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
-    from typing import ParamSpec
-else:  # pragma: no cover
+if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
     from typing_extensions import ParamSpec
 
+    t.ParamSpec = ParamSpec
+
 __all__ = ["BackgroundTask", "BackgroundTasks", "Concurrency", "BackgroundThreadTask", "BackgroundProcessTask"]
 
-P = ParamSpec("P")
+P = t.ParamSpec("P")
 
 
 class Concurrency(enum.Enum):

flama/concurrency.py

@@ -5,18 +5,19 @@
 
 from starlette.concurrency import run_in_threadpool
 
-if sys.version_info >= (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
-    from typing import ParamSpec, TypeGuard
-else:  # pragma: no cover
+if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
     from typing_extensions import ParamSpec, TypeGuard
 
+    t.TypeGuard = TypeGuard
+    t.ParamSpec = ParamSpec
+
 __all__ = ["is_async", "run"]
 
 T = t.TypeVar("T", covariant=True)
-P = ParamSpec("P")
+P = t.ParamSpec("P")
 
 
-def is_async(obj: t.Any) -> TypeGuard[t.Callable[..., t.Awaitable]]:
+def is_async(obj: t.Any) -> t.TypeGuard[t.Callable[..., t.Awaitable]]:
     """Check if given object is an async function, callable or partialised function.
 
     :param obj: Object to check.

flama/routing.py

@@ -10,10 +10,10 @@
 from flama.lifespan import Lifespan
 from flama.schemas.routing import RouteParametersMixin
 
-if sys.version_info >= (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
+if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
     from typing import TypeGuard
-else:  # pragma: no cover
-    from typing_extensions import TypeGuard
+
+    t.TypeGuard = TypeGuard
 
 if t.TYPE_CHECKING:
     from flama.applications import Flama
@@ -326,7 +326,7 @@ def __repr__(self) -> str:
     @staticmethod
     def is_endpoint(
         x: t.Union[t.Callable, t.Type[endpoints.HTTPEndpoint]]
-    ) -> TypeGuard[t.Type[endpoints.HTTPEndpoint]]:
+    ) -> t.TypeGuard[t.Type[endpoints.HTTPEndpoint]]:
         return inspect.isclass(x) and issubclass(x, endpoints.HTTPEndpoint)
 
     def endpoint_handlers(self) -> t.Dict[str, t.Callable]:
@@ -399,7 +399,7 @@ def __eq__(self, other: t.Any) -> bool:
     @staticmethod
     def is_endpoint(
         x: t.Union[t.Callable, t.Type[endpoints.WebSocketEndpoint]]
-    ) -> TypeGuard[t.Type[endpoints.WebSocketEndpoint]]:
+    ) -> t.TypeGuard[t.Type[endpoints.WebSocketEndpoint]]:
         return inspect.isclass(x) and issubclass(x, endpoints.WebSocketEndpoint)
 
     def endpoint_handlers(self) -> t.Dict[str, t.Callable]:

flama/schemas/_libs/marshmallow/adapter.py

@@ -12,11 +12,11 @@
 from flama.schemas.exceptions import SchemaGenerationError, SchemaValidationError
 from flama.schemas.types import JSONSchema
 
-if sys.version_info >= (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
-    from typing import TypeGuard
-else:  # pragma: no cover
+if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
     from typing_extensions import TypeGuard
 
+    t.TypeGuard = TypeGuard
+
 if t.TYPE_CHECKING:
     from apispec.ext.marshmallow import OpenAPIConverter
 
@@ -127,8 +127,8 @@ def unique_schema(self, schema: t.Union[Schema, t.Type[Schema]]) -> t.Type[Schem
 
         return schema
 
-    def is_schema(self, obj: t.Any) -> TypeGuard[t.Union[Schema, t.Type[Schema]]]:
+    def is_schema(self, obj: t.Any) -> t.TypeGuard[t.Union[Schema, t.Type[Schema]]]:
         return isinstance(obj, Schema) or (inspect.isclass(obj) and issubclass(obj, Schema))
 
-    def is_field(self, obj: t.Any) -> TypeGuard[t.Union[Field, t.Type[Field]]]:
+    def is_field(self, obj: t.Any) -> t.TypeGuard[t.Union[Field, t.Type[Field]]]:
         return isinstance(obj, Field) or (inspect.isclass(obj) and issubclass(obj, Field))

flama/schemas/_libs/pydantic/adapter.py

@@ -11,11 +11,11 @@
 from flama.schemas.exceptions import SchemaGenerationError, SchemaValidationError
 from flama.schemas.types import JSONSchema
 
-if sys.version_info >= (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
-    from typing import TypeGuard
-else:  # pragma: no cover
+if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
     from typing_extensions import TypeGuard
 
+    t.TypeGuard = TypeGuard
+
 __all__ = ["PydanticAdapter"]
 
 Schema = pydantic.BaseModel
@@ -111,8 +111,8 @@ def to_json_schema(self, schema: t.Union[Schema, t.Type[Schema], Field]) -> JSON
     def unique_schema(self, schema: t.Union[Schema, t.Type[Schema]]) -> t.Type[Schema]:
         return schema.__class__ if isinstance(schema, Schema) else schema
 
-    def is_schema(self, obj: t.Any) -> TypeGuard[t.Type[Schema]]:
+    def is_schema(self, obj: t.Any) -> t.TypeGuard[t.Type[Schema]]:
         return inspect.isclass(obj) and issubclass(obj, Schema)
 
-    def is_field(self, obj: t.Any) -> TypeGuard[Field]:
+    def is_field(self, obj: t.Any) -> t.TypeGuard[Field]:
         return isinstance(obj, Field)

flama/schemas/_libs/typesystem/adapter.py

@@ -10,11 +10,11 @@
 from flama.schemas.exceptions import SchemaGenerationError, SchemaValidationError
 from flama.schemas.types import JSONSchema
 
-if sys.version_info >= (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
-    from typing import TypeGuard
-else:  # pragma: no cover
+if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
     from typing_extensions import TypeGuard
 
+    t.TypeGuard = TypeGuard
+
 __all__ = ["TypesystemAdapter"]
 
 Schema = typesystem.Schema
@@ -101,9 +101,9 @@ def unique_schema(self, schema: Schema) -> Schema:
         return schema
 
     @t.no_type_check
-    def is_schema(self, obj: t.Any) -> TypeGuard[Schema]:
+    def is_schema(self, obj: t.Any) -> t.TypeGuard[Schema]:
         return isinstance(obj, Schema) or (inspect.isclass(obj) and issubclass(obj, Schema))
 
     @t.no_type_check
-    def is_field(self, obj: t.Any) -> TypeGuard[Field]:
+    def is_field(self, obj: t.Any) -> t.TypeGuard[Field]:
         return isinstance(obj, Field) or (inspect.isclass(obj) and issubclass(obj, Field))

flama/schemas/adapter.py

@@ -4,11 +4,11 @@
 
 from flama.schemas.types import Field, JSONSchema, Schema
 
-if sys.version_info >= (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
-    from typing import TypeGuard
-else:  # pragma: no cover
+if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
     from typing_extensions import TypeGuard
 
+    t.TypeGuard = TypeGuard
+
 
 class Adapter(t.Generic[Schema, Field], metaclass=abc.ABCMeta):
     DEFAULT_SCHEMA_NAME = "Schema"
@@ -79,9 +79,9 @@ def unique_schema(self, schema: t.Union[Schema, t.Type[Schema]]) -> t.Union[Sche
         ...
 
     @abc.abstractmethod
-    def is_schema(self, obj: t.Any) -> TypeGuard[t.Union[Schema, t.Type[Schema]]]:
+    def is_schema(self, obj: t.Any) -> t.TypeGuard[t.Union[Schema, t.Type[Schema]]]:
         ...
 
     @abc.abstractmethod
-    def is_field(self, obj: t.Any) -> TypeGuard[t.Union[Field, t.Type[Field]]]:
+    def is_field(self, obj: t.Any) -> t.TypeGuard[t.Union[Field, t.Type[Field]]]:
         ...

flama/schemas/data_structures.py

@@ -6,11 +6,17 @@
 from flama.injection.resolver import Parameter as InjectionParameter
 from flama.schemas.types import ParameterLocation
 
-if sys.version_info >= (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
-    from typing import TypeGuard
-else:  # pragma: no cover
+if sys.version_info < (3, 8):  # PORT: Remove when stop supporting 3.7 # pragma: no cover
+    from typing_extensions import get_args, get_origin
+
+    t.get_args = get_args
+    t.get_origin = get_origin
+
+if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
     from typing_extensions import TypeGuard
 
+    t.TypeGuard = TypeGuard
+
 __all__ = ["Field", "Schema", "Parameter", "Parameters"]
 
 
@@ -187,11 +193,11 @@ def __post_init__(self) -> None:
         object.__setattr__(self, "field", field)
 
     @property
-    def is_field(self) -> TypeGuard[Field]:
+    def is_field(self) -> t.TypeGuard[Field]:
         return isinstance(self.schema, Field)
 
     @property
-    def is_schema(self) -> TypeGuard[Schema]:
+    def is_schema(self) -> t.TypeGuard[Schema]:
         return isinstance(self.schema, Schema)
 
     @classmethod

flama/types/asgi.py

@@ -1,16 +1,17 @@
 import sys
 import typing as t
 
-if sys.version_info >= (3, 8):  # PORT: Remove when stop supporting 3.8 # pragma: no cover
-    from typing import Protocol
-else:  # pragma: no cover
+if sys.version_info < (3, 8):  # PORT: Remove when stop supporting 3.7 # pragma: no cover
     from typing_extensions import Protocol
 
-if sys.version_info >= (3, 10):  # PORT: Remove when stop supporting 3.10 # pragma: no cover
-    from typing import Concatenate, ParamSpec
-else:  # pragma: no cover
+    t.Protocol = Protocol
+
+if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
     from typing_extensions import Concatenate, ParamSpec
 
+    t.Concatenate = Concatenate
+    t.ParamSpec = ParamSpec
+
 if t.TYPE_CHECKING:
     from flama import endpoints  # noqa
 
@@ -33,30 +34,30 @@
     "WebSocketHandler",
 ]
 
-P = ParamSpec("P")
+P = t.ParamSpec("P")
 R = t.TypeVar("R", covariant=True)
 
 Scope = t.NewType("Scope", t.MutableMapping[str, t.Any])
 Message = t.NewType("Message", t.MutableMapping[str, t.Any])
 
 
-class Receive(Protocol):
+class Receive(t.Protocol):
     async def __call__(self) -> Message:
         ...
 
 
-class Send(Protocol):
+class Send(t.Protocol):
     async def __call__(self, message: Message) -> None:
         ...
 
 
 # Applications
-class AppClass(Protocol[P, R]):
+class AppClass(t.Protocol[P, R]):
     def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
         ...
 
 
-class AppAsyncClass(Protocol[P, R]):
+class AppAsyncClass(t.Protocol[P, R]):
     async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
         ...
 
@@ -77,8 +78,8 @@ def __init__(self, app: App, *args: P.args, **kwargs: P.kwargs):
         ...
 
 
-MiddlewareFunction = t.Callable[Concatenate[App, P], App]  # type: ignore[valid-type,misc]
-MiddlewareAsyncFunction = t.Callable[Concatenate[App, P], t.Awaitable[App]]  # type: ignore[valid-type,misc]
+MiddlewareFunction = t.Callable[t.Concatenate[App, P], App]  # type: ignore[valid-type,misc]
+MiddlewareAsyncFunction = t.Callable[t.Concatenate[App, P], t.Awaitable[App]]  # type: ignore[valid-type,misc]
 Middleware = t.Union[MiddlewareClass, MiddlewareAsyncClass, MiddlewareFunction, MiddlewareAsyncFunction]
 
 HTTPHandler = t.Union[AppFunction, t.Type["endpoints.HTTPEndpoint"]]