✨ Compatibility lib for older python versions (#160)

flama/background.py

@@ -1,30 +1,14 @@
 import enum
 import functools
-import sys
 import typing as t
 
 import starlette.background
 
-from flama import concurrency
-
-if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
-    from typing_extensions import ParamSpec
-
-    t.ParamSpec = ParamSpec  # type: ignore
-
-if sys.version_info < (3, 11):  # PORT: Remove when stop supporting 3.10 # pragma: no cover
-
-    class StrEnum(str, enum.Enum):
-        @staticmethod
-        def _generate_next_value_(name, start, count, last_values):
-            return name.lower()
-
-    enum.StrEnum = StrEnum  # type: ignore
-
+from flama import compat, concurrency
 
 __all__ = ["BackgroundTask", "BackgroundTasks", "Concurrency", "BackgroundThreadTask", "BackgroundProcessTask"]
 
-P = t.ParamSpec("P")  # type: ignore # PORT: Remove this comment when stop supporting 3.9
+P = compat.ParamSpec("P")  # PORT: Replace compat when stop supporting 3.9
 
 
 class task_wrapper:
@@ -36,7 +20,7 @@ async def __call__(self, *args, **kwargs):
         await concurrency.run(self.target, *args, **kwargs)
 
 
-class Concurrency(enum.StrEnum):  # type: ignore # PORT: Remove this comment when stop supporting 3.10
+class Concurrency(compat.StrEnum):  # PORT: Replace compat when stop supporting 3.10
     thread = enum.auto()
     process = enum.auto()
 

flama/compat.py

@@ -0,0 +1,66 @@
+import sys
+
+__all__ = ["Concatenate", "ParamSpec", "TypeGuard", "UnionType", "StrEnum", "tomllib"]
+
+# PORT: Remove when stop supporting 3.9
+# Concatenate was added in Python 3.10
+# https://docs.python.org/3/library/typing.html#typing.Concatenate
+if sys.version_info >= (3, 10):
+    from typing import Concatenate
+else:
+    from typing_extensions import Concatenate
+
+
+# PORT: Remove when stop supporting 3.9
+# ParamSpec was added in Python 3.10
+# https://docs.python.org/3/library/typing.html#typing.ParamSpec
+if sys.version_info >= (3, 10):
+    from typing import ParamSpec
+else:
+    from typing_extensions import ParamSpec
+
+
+# PORT: Remove when stop supporting 3.9
+# TypeGuard was added in Python 3.10
+# https://docs.python.org/3/library/typing.html#typing.TypeGuard
+if sys.version_info >= (3, 10):
+    from typing import TypeGuard
+else:
+    from typing_extensions import TypeGuard
+
+
+# PORT: Remove when stop supporting 3.9
+# UnionType was added in Python 3.10
+# https://docs.python.org/3/library/stdtypes.html#types-union
+if sys.version_info >= (3, 10):
+    from types import UnionType
+else:
+    from typing import Union as UnionType
+
+
+# PORT: Remove when stop supporting 3.10
+# StrEnum was added in Python 3.11
+# https://docs.python.org/3/library/enum.html#enum.StrEnum
+if sys.version_info >= (3, 11):
+    from enum import StrEnum
+else:
+    from enum import Enum
+
+    class StrEnum(str, Enum):
+        @staticmethod
+        def _generate_next_value_(name, start, count, last_values):
+            return name.lower()
+
+
+# PORT: Remove when stop supporting 3.10
+# Tomllib was added in Python 3.11
+# https://docs.python.org/3/library/tomllib.html
+if sys.version_info >= (3, 11):  # PORT: Remove when stop supporting 3.10
+    import tomllib
+else:
+    try:
+        import tomli
+
+        tomllib = tomli
+    except ModuleNotFoundError:
+        tomllib = None

flama/concurrency.py

@@ -4,23 +4,17 @@
 import sys
 import typing as t
 
-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  # type: ignore
-    t.ParamSpec = ParamSpec  # type: ignore
+from flama import compat
 
 __all__ = ["is_async", "run", "run_task_group", "AsyncProcess"]
 
 R = t.TypeVar("R", covariant=True)
-P = t.ParamSpec("P")  # type: ignore # PORT: Remove this comment when stop supporting 3.9
+P = compat.ParamSpec("P")  # PORT: Replace compat when stop supporting 3.9
 
 
 def is_async(
     obj: t.Any,
-) -> t.TypeGuard[  # type: ignore # PORT: Remove this comment when stop supporting 3.9
-    t.Callable[..., t.Awaitable[t.Any]]
-]:
+) -> compat.TypeGuard[t.Callable[..., t.Awaitable[t.Any]]]:  # PORT: Replace compat when stop supporting 3.9
     """Check if given object is an async function, callable or partialised function.
 
     :param obj: Object to check.
@@ -51,28 +45,28 @@ async def run(
     return t.cast(R, await asyncio.to_thread(func, *args, **kwargs))
 
 
-if sys.version_info < (3, 11):  # PORT: Remove when stop supporting 3.10 # pragma: no cover
+if sys.version_info >= (3, 11):  # PORT: Remove when stop supporting 3.10 # pragma: no cover
 
     async def run_task_group(*tasks: t.Coroutine[t.Any, t.Any, t.Any]) -> list[asyncio.Task]:
         """Run a group of tasks.
 
         :param tasks: Tasks to run.
         :result: Finished tasks.
         """
-        tasks_list = [asyncio.create_task(task) for task in tasks]
-        await asyncio.wait(tasks_list)
-        return tasks_list
+        async with asyncio.TaskGroup() as task_group:
+            return [task_group.create_task(task) for task in tasks]
 
-else:  # noqa
+else:  # pragma: no cover
 
     async def run_task_group(*tasks: t.Coroutine[t.Any, t.Any, t.Any]) -> list[asyncio.Task]:
         """Run a group of tasks.
 
         :param tasks: Tasks to run.
         :result: Finished tasks.
         """
-        async with asyncio.TaskGroup() as task_group:
-            return [task_group.create_task(task) for task in tasks]
+        tasks_list = [asyncio.create_task(task) for task in tasks]
+        await asyncio.wait(tasks_list)
+        return tasks_list
 
 
 class AsyncProcess(multiprocessing.Process):

flama/config/loaders.py

@@ -2,22 +2,11 @@
 import configparser
 import json
 import os
-import sys
 import typing as t
 
 import yaml
 
-from flama import exceptions
-
-if sys.version_info < (3, 11):  # PORT: Remove when stop supporting 3.10 # pragma: no cover
-    try:
-        import tomli
-
-        tomllib = tomli
-    except ModuleNotFoundError:
-        tomllib = None
-else:  # noqa
-    import tomllib
+from flama import compat, exceptions
 
 __all__ = ["FileLoader", "ConfigFileLoader", "JSONFileLoader", "YAMLFileLoader", "TOMLFileLoader"]
 
@@ -90,12 +79,12 @@ def load(self, f: t.Union[str, os.PathLike]) -> dict[str, t.Any]:
         :param f: File path.
         :return: Dict with the file contents.
         """
-        if tomllib is None:
+        if compat.tomllib is None:  # PORT: Replace compat when stop supporting 3.10
             raise exceptions.DependencyNotInstalled(
                 dependency=exceptions.DependencyNotInstalled.Dependency.tomli,
                 dependant=f"{self.__class__.__module__}.{self.__class__.__name__}",
                 msg="for Python versions lower than 3.11",
             )
 
         with open(f, "rb") as fs:
-            return tomllib.load(fs)
+            return compat.tomllib.load(fs)  # PORT: Replace compat when stop supporting 3.10

flama/config/types.py

@@ -1,19 +1,11 @@
 import enum
-import sys
 
-__all__ = ["FileFormat"]
-
-if sys.version_info < (3, 11):  # PORT: Remove when stop supporting 3.10 # pragma: no cover
+from flama import compat
 
-    class StrEnum(str, enum.Enum):
-        @staticmethod
-        def _generate_next_value_(name, start, count, last_values):
-            return name.lower()
-
-    enum.StrEnum = StrEnum  # type: ignore
+__all__ = ["FileFormat"]
 
 
-class FileFormat(enum.StrEnum):  # type: ignore # PORT: Remove this comment when stop supporting 3.10
+class FileFormat(compat.StrEnum):  # PORT: Replace compat when stop supporting 3.10
     """Config file format."""
 
     ini = enum.auto()

flama/exceptions.py

@@ -1,10 +1,10 @@
-import enum
 import http
-import sys
 import typing as t
 
 import starlette.exceptions
 
+from flama import compat
+
 __all__ = [
     "ApplicationError",
     "DependencyNotInstalled",
@@ -21,22 +21,13 @@
     "FrameworkVersionWarning",
 ]
 
-if sys.version_info < (3, 11):  # PORT: Remove when stop supporting 3.10 # pragma: no cover
-
-    class StrEnum(str, enum.Enum):
-        @staticmethod
-        def _generate_next_value_(name, start, count, last_values):
-            return name.lower()
-
-    enum.StrEnum = StrEnum  # type: ignore
-
 
 class ApplicationError(Exception):
     ...
 
 
 class DependencyNotInstalled(ApplicationError):
-    class Dependency(enum.StrEnum):  # type: ignore # PORT: Remove this comment when stop supporting 3.10
+    class Dependency(compat.StrEnum):  # PORT: Replace compat when stop supporting 3.10
         pydantic = "pydantic"
         marshmallow = "marshmallow"
         apispec = "apispec"

flama/http.py

@@ -5,7 +5,6 @@
 import inspect
 import json
 import os
-import sys
 import typing as t
 import uuid
 from pathlib import Path
@@ -15,16 +14,7 @@
 import starlette.responses
 import starlette.schemas
 
-from flama import exceptions, schemas, types
-
-if sys.version_info < (3, 11):  # PORT: Remove when stop supporting 3.10 # pragma: no cover
-
-    class StrEnum(str, enum.Enum):
-        @staticmethod
-        def _generate_next_value_(name, start, count, last_values):
-            return name.lower()
-
-    enum.StrEnum = StrEnum  # type: ignore
+from flama import compat, exceptions, schemas, types
 
 __all__ = [
     "Method",
@@ -43,7 +33,7 @@ def _generate_next_value_(name, start, count, last_values):
     "OpenAPIResponse",
 ]
 
-Method = enum.StrEnum(  # type: ignore # PORT: Remove this comment when stop supporting 3.10
+Method = compat.StrEnum(  # PORT: Replace compat when stop supporting 3.10
     "Method", ["GET", "HEAD", "POST", "PUT", "DELETE", "CONNECT", "OPTIONS", "TRACE", "PATCH"]
 )
 Request = starlette.requests.Request

flama/pagination/types.py

@@ -1,18 +1,10 @@
 import enum
-import sys
 
-if sys.version_info < (3, 11):  # PORT: Remove when stop supporting 3.10 # pragma: no cover
-
-    class StrEnum(str, enum.Enum):
-        @staticmethod
-        def _generate_next_value_(name, start, count, last_values):
-            return name.lower()
-
-    enum.StrEnum = StrEnum  # type: ignore
+from flama import compat
 
 __all__ = ["PaginationType"]
 
 
-class PaginationType(enum.StrEnum):  # type: ignore # PORT: Remove this comment when stop supporting 3.10
+class PaginationType(compat.StrEnum):  # PORT: Replace compat when stop supporting 3.10
     page_number = enum.auto()
     limit_offset = enum.auto()

flama/routing.py

@@ -4,29 +4,14 @@
 import functools
 import inspect
 import logging
-import sys
 import typing as t
 
-from flama import concurrency, endpoints, exceptions, http, schemas, types, url, websockets
+from flama import compat, concurrency, endpoints, exceptions, http, schemas, types, url, websockets
 from flama.injection import Component, Components
 from flama.lifespan import Lifespan
 from flama.pagination import paginator
 from flama.schemas.routing import RouteParametersMixin
 
-if sys.version_info < (3, 10):  # PORT: Remove when stop supporting 3.9 # pragma: no cover
-    from typing_extensions import TypeGuard
-
-    t.TypeGuard = TypeGuard  # type: ignore
-
-if sys.version_info < (3, 11):  # PORT: Remove when stop supporting 3.10 # pragma: no cover
-
-    class StrEnum(str, enum.Enum):
-        @staticmethod
-        def _generate_next_value_(name, start, count, last_values):
-            return name.lower()
-
-    enum.StrEnum = StrEnum  # type: ignore
-
 if t.TYPE_CHECKING:
     from flama.applications import Flama
     from flama.pagination.types import PaginationType
@@ -36,7 +21,7 @@ def _generate_next_value_(name, start, count, last_values):
 logger = logging.getLogger(__name__)
 
 
-class _EndpointType(enum.StrEnum):  # type: ignore # PORT: Remove this comment when stop supporting 3.10
+class _EndpointType(compat.StrEnum):  # PORT: Replace compat when stop supporting 3.10
     http = enum.auto()
     websocket = enum.auto()
 
@@ -363,7 +348,7 @@ def __repr__(self) -> str:
     @staticmethod
     def is_endpoint(
         x: t.Union[t.Callable, type[endpoints.HTTPEndpoint]]
-    ) -> t.TypeGuard[type[endpoints.HTTPEndpoint]]:  # type: ignore # PORT: Remove this comment when stop supporting 3.9
+    ) -> compat.TypeGuard[type[endpoints.HTTPEndpoint]]:  # PORT: Replace compat when stop supporting 3.9
         return inspect.isclass(x) and issubclass(x, endpoints.HTTPEndpoint)
 
     def endpoint_handlers(self) -> dict[str, t.Callable]:
@@ -445,9 +430,7 @@ def __eq__(self, other: t.Any) -> bool:
     @staticmethod
     def is_endpoint(
         x: t.Union[t.Callable, type[endpoints.WebSocketEndpoint]]
-    ) -> t.TypeGuard[  # type: ignore # PORT: Remove this comment when stop supporting 3.9
-        type[endpoints.WebSocketEndpoint]
-    ]:
+    ) -> compat.TypeGuard[type[endpoints.WebSocketEndpoint]]:  # PORT: Replace compat when stop supporting 3.9
         return inspect.isclass(x) and issubclass(x, endpoints.WebSocketEndpoint)
 
     def endpoint_handlers(self) -> dict[str, t.Callable]: