aio.core: Improve typing (#2331)

Signed-off-by: Ryan Northey <ryan@synca.io>

aio.core/VERSION

@@ -1 +1 @@
-0.10.4-dev
+0.10.4

aio.core/aio/core/BUILD

@@ -6,6 +6,9 @@ toolshed_library(
         "//deps:reqs#pytz",
         "//deps:reqs#pyyaml",
         "//deps:reqs#trycast",
+        "//deps:reqs#types-orjson",
+        "//deps:reqs#types-pytz",
+        "//deps:reqs#types-pyyaml",
     ],
     sources=[
         "__init__.py",

aio.core/aio/core/dev/debug.py

@@ -4,7 +4,7 @@
 import os
 import time
 from functools import cached_property, partial
-from typing import Any, Callable, Iterator, Optional
+from typing import Any, Callable, Iterator
 
 
 logger = _logging.getLogger(__name__)
@@ -14,7 +14,7 @@ class ADebugLogging:
 
     def __init__(
             self,
-            fun: Optional[Callable] = None,
+            fun: Callable | None = None,
             *args,
             **kwargs) -> None:
         self.__wrapped__ = fun
@@ -23,7 +23,7 @@ def __init__(
         self._format_result = kwargs.pop("format_result", None)
         self._show_cpu = kwargs.pop("show_cpu", None)
 
-    def __call__(self, *args, **kwargs) -> Optional[Any]:
+    def __call__(self, *args, **kwargs) -> Any:
         if self.__wrapped__:
             if inspect.isasyncgenfunction(self.__wrapped__):
                 return self.fun_async_gen(*args, **kwargs)
@@ -32,7 +32,7 @@ def __call__(self, *args, **kwargs) -> Optional[Any]:
             elif inspect.isgeneratorfunction(self.__wrapped__):
                 return self.fun_gen(*args, **kwargs)
             return self.fun(*args, **kwargs)
-        fun: Optional[Callable] = args[0] if args else None
+        fun: Callable | None = args[0] if args else None
         self.__wrapped__ = fun
         self.__doc__ = getattr(fun, '__doc__')
         return self
@@ -66,15 +66,15 @@ def log(self, instance) -> _logging.Logger:
             return self._log
         return logger
 
-    def format_result(self, instance) -> Optional[Callable]:
+    def format_result(self, instance) -> Callable | None:
         if not self._format_result:
             return None
         if isinstance(self._format_result, str):
             if self._format_result.startswith("self."):
                 return getattr(instance, self._format_result[5:])
         return self._format_result
 
-    def fun(self, *args, **kwargs) -> Optional[Any]:
+    def fun(self, *args, **kwargs) -> Any:
         if not callable(self.__wrapped__):
             return None
         return self.log_debug_complete(

aio.core/aio/core/directory/abstract/directory.py

@@ -7,8 +7,8 @@
 from concurrent import futures
 from functools import cached_property, partial
 from typing import (
-    AsyncIterator, Dict, Iterable, List, Mapping, Optional,
-    Pattern, Set, Tuple, Type, Union)
+    AsyncIterator, Iterable, Mapping,
+    Pattern, Type)
 
 import abstracts
 
@@ -42,13 +42,13 @@ def include_path(cls, path: str, path_matcher, exclude_matcher) -> bool:
     def __init__(
             self,
             path: str,
-            path_matcher: Optional[Pattern[str]] = None,
-            exclude_matcher:  Optional[Pattern[str]] = None) -> None:
+            path_matcher: Pattern[str] | None = None,
+            exclude_matcher:  Pattern[str] | None = None) -> None:
         _subprocess.ASubprocessHandler.__init__(self, path)
         self.path_matcher = path_matcher
         self.exclude_matcher = exclude_matcher
 
-    def handle(self, response: subprocess.CompletedProcess) -> Set[str]:
+    def handle(self, response: subprocess.CompletedProcess) -> set[str]:
         return set(
             path
             for path
@@ -60,7 +60,7 @@ def handle(self, response: subprocess.CompletedProcess) -> Set[str]:
 
     def handle_error(
             self,
-            response: subprocess.CompletedProcess) -> Dict[str, List[str]]:
+            response: subprocess.CompletedProcess) -> dict[str, list[str]]:
         # TODO: Handle errors in directory classes
         return super().handle_error(response)
 
@@ -79,8 +79,8 @@ class AGitDirectoryFileFinder(
     def __init__(
             self,
             path: str,
-            path_matcher: Optional[Pattern[str]] = None,
-            exclude_matcher:  Optional[Pattern[str]] = None,
+            path_matcher: Pattern[str] | None = None,
+            exclude_matcher:  Pattern[str] | None = None,
             match_binaries: bool = False,
             match_all_files: bool = False) -> None:
         self.match_binaries = match_binaries
@@ -105,12 +105,12 @@ def parse_response(
                 files.append(filename)
         return files
 
-    def _get_file(self, line: str) -> Optional[str]:
+    def _get_file(self, line: str) -> str | None:
         eol, name = self._parse_line(line)
         if eol and (self.match_binaries or eol not in ["-text", "none"]):
             return name
 
-    def _parse_line(self, line: str) -> Tuple[Optional[str], Optional[str]]:
+    def _parse_line(self, line: str) -> tuple[str | None, str | None]:
         matched = self.matcher.match(line)
         return (
             matched.groups()
@@ -138,14 +138,14 @@ class ADirectory(event.AExecutive, metaclass=abstracts.Abstraction):
 
     def __init__(
             self,
-            path: Union[str, pathlib.Path],
-            exclude: Optional[Iterable[str]] = None,
-            exclude_dirs: Optional[Iterable[str]] = None,
-            path_matcher: Optional[Pattern[str]] = None,
-            exclude_matcher: Optional[Pattern[str]] = None,
-            text_only: Optional[bool] = True,
-            loop: Optional[asyncio.AbstractEventLoop] = None,
-            pool: Optional[futures.Executor] = None) -> None:
+            path: str | pathlib.Path,
+            exclude: Iterable[str] | None = None,
+            exclude_dirs: Iterable[str] | None = None,
+            path_matcher: Pattern[str] | None = None,
+            exclude_matcher: Pattern[str] | None = None,
+            text_only: bool | None = True,
+            loop: asyncio.AbstractEventLoop | None = None,
+            pool: futures.Executor | None = None) -> None:
         self._path = path
         self.path_matcher = path_matcher
         self.exclude_matcher = exclude_matcher
@@ -160,8 +160,8 @@ def absolute_path(self) -> str:
         return str(self.path.resolve())
 
     @async_property(cache=True)
-    async def files(self) -> Set[str]:
-        """Set of relative file paths associated with this directory."""
+    async def files(self) -> set[str]:
+        """set of relative file paths associated with this directory."""
         return await self.get_files()
 
     @cached_property
@@ -182,7 +182,7 @@ def finder_class(self) -> Type[ADirectoryFileFinder]:
         raise NotImplementedError
 
     @cached_property
-    def grep_args(self) -> Tuple[str, ...]:
+    def grep_args(self) -> tuple[str, ...]:
         """Default args to pass when calling `grep`."""
         return (
             *self.grep_command_args,
@@ -192,15 +192,15 @@ def grep_args(self) -> Tuple[str, ...]:
             *self.grep_exclusion_args)
 
     @property
-    def grep_command_args(self) -> Tuple[str, ...]:
+    def grep_command_args(self) -> tuple[str, ...]:
         """Path args for the `grep` command."""
         if grep_command := shutil.which("grep"):
             return grep_command, "-r"
         raise _subprocess.exceptions.OSCommandError(
             "Unable to find `grep` command")
 
     @property
-    def grep_exclusion_args(self) -> Tuple[str, ...]:
+    def grep_exclusion_args(self) -> tuple[str, ...]:
         """Grep flags to exclude paths and directories."""
         return (
             tuple(
@@ -240,23 +240,23 @@ def filtered(self, **kwargs):
         init_kwargs.update(kwargs)
         return self.__class__(**init_kwargs)
 
-    async def get_files(self) -> Set[str]:
+    async def get_files(self) -> set[str]:
         return await self.grep(["-l"], "")
 
     def grep(
             self,
             args: Iterable,
-            target: Union[str, Iterable[str]]) -> AwaitableGenerator:
+            target: str | Iterable[str]) -> AwaitableGenerator:
         return AwaitableGenerator(
             self._grep(args, target),
             collector=async_set)
 
     def parse_grep_args(
             self,
             args: Iterable[str],
-            target: Union[
-                str,
-                Iterable[str]]) -> Tuple[Tuple[str, ...], Iterable[str]]:
+            target: (
+                str
+                | Iterable[str])) -> tuple[tuple[str, ...], Iterable[str]]:
         """Parse `grep` args with the defaults to pass to the `grep`
         command."""
         return (
@@ -268,7 +268,7 @@ def parse_grep_args(
 
     def _batched_grep(
             self,
-            grep_args: Tuple[str, ...],
+            grep_args: tuple[str, ...],
             paths: Iterable[str]) -> AwaitableGenerator:
         return self.execute_in_batches(
             partial(self.finder, *grep_args),
@@ -279,7 +279,7 @@ def _batched_grep(
     async def _grep(
             self,
             args: Iterable[str],
-            target: Union[str, Iterable[str]]) -> AsyncIterator[str]:
+            target: str | Iterable[str]) -> AsyncIterator[str]:
         """Run `grep` in the directory."""
         if not isinstance(target, str) and not target:
             return
@@ -301,7 +301,7 @@ class AGitDirectory(ADirectory):
     def find_git_deleted_files(
             cls,
             finder: AGitDirectoryFileFinder,
-            git_command: str) -> Set[str]:
+            git_command: str) -> set[str]:
         return finder(
             git_command,
             "ls-files",
@@ -312,7 +312,7 @@ def find_git_files_changed_since(
             cls,
             finder: AGitDirectoryFileFinder,
             git_command: str,
-            since: str) -> Set[str]:
+            since: str) -> set[str]:
         return finder(
             git_command,
             "diff",
@@ -325,7 +325,7 @@ def find_git_files_changed_since(
     def find_git_untracked_files(
             cls,
             finder: AGitDirectoryFileFinder,
-            git_command: str) -> Set[str]:
+            git_command: str) -> set[str]:
         return finder(
             git_command,
             "ls-files",
@@ -340,7 +340,7 @@ def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
 
     @async_property(cache=True)
-    async def changed_files(self) -> Set[str]:
+    async def changed_files(self) -> set[str]:
         """Files that have changed since `self.changed`, which can be the name
         of a git object (branch etc) or a commit hash."""
         return await self.execute(
@@ -350,7 +350,7 @@ async def changed_files(self) -> Set[str]:
             self.changed)
 
     @async_property(cache=True)
-    async def deleted_files(self) -> Set[str]:
+    async def deleted_files(self) -> set[str]:
         """Files that have changed since `self.changed`, which can be the name
         of a git object (branch etc) or a commit hash."""
         return await self.execute(
@@ -359,7 +359,7 @@ async def deleted_files(self) -> Set[str]:
             self.git_command)
 
     @async_property(cache=True)
-    async def files(self) -> Set[str]:
+    async def files(self) -> set[str]:
         """Files that have changed."""
         if not self.changed:
             return await self.get_files()
@@ -394,7 +394,7 @@ def git_command(self) -> str:
             "Unable to find the `git` command")
 
     @property
-    def grep_command_args(self) -> Tuple[str, ...]:
+    def grep_command_args(self) -> tuple[str, ...]:
         return self.git_command, "grep", "--cached"
 
     @property
@@ -408,7 +408,7 @@ async def untracked_files(self):
             self.finder,
             self.git_command)
 
-    async def get_files(self) -> Set[str]:
+    async def get_files(self) -> set[str]:
         return (
             await super().get_files() - await self.deleted_files
             if not self.untracked

aio.core/aio/core/directory/context.py

@@ -2,7 +2,6 @@
 import os
 import pathlib
 from functools import cached_property
-from typing import Optional, Union
 
 import abstracts
 
@@ -26,11 +25,11 @@ def in_directory(self):
 
 @abstracts.implementer(IDirectoryContext)
 class ADirectoryContext(metaclass=abstracts.Abstraction):
-    _path: Optional[Union[str, os.PathLike]] = None
+    _path: str | os.PathLike | None = None
 
     def __init__(
             self,
-            path: Union[str, os.PathLike]) -> None:
+            path: str | os.PathLike) -> None:
         self._path = path
 
     @cached_property

aio.core/aio/core/directory/utils.py

@@ -2,11 +2,10 @@
 import os
 import pathlib
 from contextlib import contextmanager
-from typing import Union
 
 
 @contextmanager
-def directory_context(path: Union[str, pathlib.Path]):
+def directory_context(path: str | pathlib.Path):
     """Sets the directory context."""
     origin = pathlib.Path().absolute()
     try:

aio.core/aio/core/event/executive.py

@@ -1,6 +1,6 @@
 
 from functools import partial
-from typing import Any, Callable, Optional
+from typing import Any, Callable
 
 import abstracts
 
@@ -26,9 +26,9 @@ async def execute_in_batches(
             self,
             executable: Callable,
             *args,
-            concurrency: Optional[int] = None,
-            min_batch_size: Optional[int] = None,
-            max_batch_size: Optional[int] = None,
+            concurrency: int | None = None,
+            min_batch_size: int | None = None,
+            max_batch_size: int | None = None,
             **kwargs) -> functional.AwaitableGenerator:
         """Execute a command in a process pool."""
         raise NotImplementedError
@@ -54,9 +54,9 @@ def execute_in_batches(
             self,
             executable: Callable,
             *args,
-            concurrency: Optional[int] = None,
-            min_batch_size: Optional[int] = None,
-            max_batch_size: Optional[int] = None,
+            concurrency: int | None = None,
+            min_batch_size: int | None = None,
+            max_batch_size: int | None = None,
             **kwargs) -> functional.AwaitableGenerator:
         return tasks.concurrent(
             (self.execute(

aio.core/aio/core/event/reactive.py

@@ -2,7 +2,6 @@
 import asyncio
 from concurrent import futures
 from functools import cached_property
-from typing import Optional
 
 import abstracts
 
@@ -24,8 +23,8 @@ def pool(self) -> futures.Executor:
 
 
 class AReactive(IReactive, metaclass=abstracts.Abstraction):
-    _loop: Optional[asyncio.AbstractEventLoop] = None
-    _pool: Optional[futures.Executor] = None
+    _loop: asyncio.AbstractEventLoop | None = None
+    _pool: futures.Executor | None = None
 
     @cached_property
     def loop(self) -> asyncio.AbstractEventLoop: