feat: add type annotations and method signatures to storage and typesf modules

Author: joamag

src/appier/storage.py

@@ -53,19 +53,19 @@ def read(cls, file, *args, **kwargs):
         raise exceptions.NotImplementedError()
 
     @classmethod
-    def seek(self, file, *args, **kwargs):
+    def seek(cls, file, *args, **kwargs):
         raise exceptions.NotImplementedError()
 
     @classmethod
-    def cleanup(self, file, *args, **kwargs):
+    def cleanup(cls, file, *args, **kwargs):
         raise exceptions.NotImplementedError()
 
     @classmethod
-    def is_seekable(self):
+    def is_seekable(cls):
         return False
 
     @classmethod
-    def is_stored(self):
+    def is_stored(cls):
         return False
 
     @classmethod
@@ -75,7 +75,9 @@ def _compute(cls, file, *args, **kwargs):
 
 class BaseEngine(StorageEngine):
     @classmethod
-    def load(cls, file, *args, **kwargs):
+    def load(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         force = kwargs.get("force", False)
         if not file.file_name:
             return
@@ -92,18 +94,24 @@ def load(cls, file, *args, **kwargs):
         finally:
             handle.close()
 
-        cls._compute()
+        cls._compute(file)
 
     @classmethod
-    def store(cls, file, *args, **kwargs):
+    def store(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         pass
 
     @classmethod
-    def delete(cls, file, *args, **kwargs):
+    def delete(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         pass
 
     @classmethod
-    def read(cls, file, *args, **kwargs):
+    def read(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         # tries to determine the requested size for# the file
         # reading in case none is defined the handled flag
         # handling is ignored and the data returned immediately
@@ -125,19 +133,23 @@ def read(cls, file, *args, **kwargs):
         return None
 
     @classmethod
-    def cleanup(cls, file, *args, **kwargs):
+    def cleanup(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         if not hasattr(file, "handled"):
             return
         del file._handled
 
     @classmethod
-    def is_stored(self):
+    def is_stored(cls):  # pyright: ignore[reportIncompatibleMethodOverride]
         return True
 
 
 class FsEngine(StorageEngine):
     @classmethod
-    def store(cls, file, *args, **kwargs):
+    def store(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         file_path = cls._file_path(file)
         file_data = file.data or b""
         handle = open(file_path, "wb")
@@ -147,43 +159,54 @@ def store(cls, file, *args, **kwargs):
             handle.close()
 
     @classmethod
-    def load(cls, file, *args, **kwargs):
+    def load(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         cls._compute(file)
 
     @classmethod
-    def delete(cls, file, *args, **kwargs):
+    def delete(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         file_path = cls._file_path(file, ensure=False)
         os.remove(file_path)
 
     @classmethod
-    def read(cls, file, *args, **kwargs):
+    def read(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         data = None
         size = kwargs.get("size", None)
         handle = cls._handle(file)
         try:
             data = handle.read(size or -1)
         finally:
             is_final = True if not size or not data else False
-            is_final and cls.cleanup(file)
+            if is_final:
+                cls.cleanup(file)
         return data
 
     @classmethod
-    def seek(cls, file, *args, **kwargs):
+    def seek(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         offset = kwargs.get("offset", None)
         if offset == None:
             return
         handle = cls._handle(file)
         handle.seek(offset)
 
     @classmethod
-    def cleanup(cls, file, *args, **kwargs):
+    def cleanup(  # pyright: ignore[reportIncompatibleMethodOverride]
+        cls, file, *args, **kwargs
+    ):
         if not hasattr(file, "_handle"):
             return
         file._handle.close()
         del file._handle
 
     @classmethod
-    def is_seekable(self):
+    def is_seekable(cls):  # pyright: ignore[reportIncompatibleMethodOverride]
         return True
 
     @classmethod
@@ -207,7 +230,7 @@ def _handle(cls, file):
     @classmethod
     def _file_path(cls, file, ensure=True, base=None):
         # verifies that the standard params value is defined and
-        # if that's no the case defaults the value, then tries to
+        # if that's not the case defaults the value, then tries to
         # retrieve a series of parameters for file path discovery
         params = file.params or {}
         file_path = params.get("file_path", None)

src/appier/storage.pyi

@@ -0,0 +1,61 @@
+from typing import Any
+
+from .typesf import File
+
+class StorageEngine:
+    @classmethod
+    def load(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def store(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def delete(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def read(cls, file: File, *args, **kwargs) -> bytes | None: ...
+    @classmethod
+    def seek(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def cleanup(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def is_seekable(cls) -> bool: ...
+    @classmethod
+    def is_stored(cls) -> bool: ...
+    @classmethod
+    def _compute(cls, file: File, *args, **kwargs) -> None: ...
+
+class BaseEngine(StorageEngine):
+    @classmethod
+    def load(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def store(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def delete(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def read(cls, file: File, *args, **kwargs) -> bytes | None: ...
+    @classmethod
+    def cleanup(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def is_stored(cls) -> bool: ...
+
+class FsEngine(StorageEngine):
+    @classmethod
+    def store(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def load(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def delete(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def read(cls, file: File, *args, **kwargs) -> bytes | None: ...
+    @classmethod
+    def seek(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def cleanup(cls, file: File, *args, **kwargs) -> None: ...
+    @classmethod
+    def is_seekable(cls) -> bool: ...
+    @classmethod
+    def _compute(cls, file: File) -> None: ...
+    @classmethod
+    def _handle(cls, file: File) -> Any: ...
+    @classmethod
+    def _file_path(
+        cls, file: File, ensure: bool = ..., base: str | None = ...
+    ) -> str: ...

src/appier/typesf.pyi

@@ -0,0 +1,102 @@
+from typing import Any, Callable, IO, Iterator
+
+class AbstractType:
+    def json_v(self, *args, **kwargs) -> Any: ...
+    def map_v(self, *args, **kwargs) -> Any: ...
+
+class Type(AbstractType):
+    def __init__(self, value: Any): ...
+    def json_v(self, *args, **kwargs) -> Any: ...
+    def loads(self, value: Any) -> None: ...
+    def dumps(self) -> Any: ...
+
+class File(AbstractType):
+    data: bytes | None
+    data_b64: str | None
+    file: Any | None
+    hash: str | None
+    size: int
+    file_name: str
+    mime: str | None
+    etag: str | None
+    guid: str
+    params: dict[str, Any] | None
+    engine: str | None
+    def __init__(self, file: Any): ...
+    def __len__(self) -> int: ...
+    @classmethod
+    def open(cls, path: str) -> File: ...
+    def build_d(self, file_d: bytes, name: str = ...) -> None: ...
+    def build_b64(self, file_m: dict[str, Any]) -> None: ...
+    def build_t(self, file_t: tuple[str, str | None, bytes | None]) -> None: ...
+    def build_i(self, file: File) -> None: ...
+    def build_f(self, file: Any) -> None: ...
+    def read(self, size: int | None = ...) -> bytes | None: ...
+    def seek(self, offset: int | None = ...) -> None: ...
+    def delete(self) -> None: ...
+    def cleanup(self) -> None: ...
+    def save(self, path: str | None = ...) -> None: ...
+    def json_v(self, *args, **kwargs) -> dict[str, Any] | None: ...
+    def is_seekable(self) -> bool: ...
+    def is_stored(self) -> bool: ...
+    def is_valid(self) -> bool: ...
+    def is_empty(self) -> bool: ...
+    def _hash(self, data: bytes | None) -> str | None: ...
+    def _etag(self, data: bytes | None) -> str | None: ...
+    def _guid(self) -> str: ...
+    def _load(self, force: bool = ...) -> None: ...
+    def _store(self, force: bool = ...) -> None: ...
+    def _compute(self) -> None: ...
+    def _engine(self) -> Any: ...
+
+class Files(AbstractType):
+    _files: list[File]
+    def __init__(self, files: Any): ...
+    def __len__(self) -> int: ...
+    def __iter__(self) -> Iterator[File]: ...
+    def __getitem__(self, key: int) -> File: ...
+    def base(self) -> type[File]: ...
+    def build_i(self, files: Files) -> None: ...
+    def build_f(self, files: Any) -> None: ...
+    def json_v(self, *args, **kwargs) -> list[dict[str, Any]]: ...
+    def is_empty(self) -> bool: ...
+    def _load(self) -> None: ...
+
+class ImageFile(File):
+    width: int
+    height: int
+    format: str | None
+    kwargs: dict[str, Any]
+    def build_b64(self, file_m: dict[str, Any]) -> None: ...
+    def build_t(self, file_t: tuple[str, str | None, bytes | None]) -> None: ...
+    def build_i(self, file: File) -> None: ...
+    def build_f(self, file: Any) -> None: ...
+    def json_v(self, *args, **kwargs) -> dict[str, Any] | None: ...
+    def _ensure_all(self) -> None: ...
+    def _ensure_size(self) -> None: ...
+    def _ensure_mime(self) -> None: ...
+    def _ensure_kwargs(self) -> None: ...
+    def _size(self) -> tuple[int, int]: ...
+    def _size_image(self) -> tuple[int, int]: ...
+    def _size_pil(self) -> tuple[int, int]: ...
+    def _size_default(self) -> tuple[int, int]: ...
+    def _mime(self) -> tuple[str | None, str]: ...
+    def _mime_image(self) -> tuple[str | None, str]: ...
+    def _mime_pil(self) -> tuple[str | None, str]: ...
+    def _mime_default(self) -> tuple[str | None, str]: ...
+
+class ImageFiles(Files):
+    def base(self) -> type[ImageFile]: ...
+
+def image(
+    width: int | None = ..., height: int | None = ..., format: str = ..., **kwargs
+) -> type[ImageFile]: ...
+def images(
+    width: int | None = ..., height: int | None = ..., format: str = ..., **kwargs
+) -> type[ImageFiles]: ...
+
+class Reference(AbstractType): ...
+
+def reference(
+    target: Any, name: str | None = ..., dumpall: bool = ...
+) -> type[Reference]: ...