venv

1 files changed, 154 insertions, 0 deletions

diff --git a/venv/lib/python3.11/site-packages/litestar/file_system.py b/venv/lib/python3.11/site-packages/litestar/file_system.py
new file mode 100644
index 0000000..fcb77c7
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/file_system.py
@@ -0,0 +1,154 @@
+from __future__ import annotations
+
+from stat import S_ISDIR
+from typing import TYPE_CHECKING, Any, AnyStr, cast
+
+from anyio import AsyncFile, Path, open_file
+
+from litestar.concurrency import sync_to_thread
+from litestar.exceptions import InternalServerException, NotAuthorizedException
+from litestar.types.file_types import FileSystemProtocol
+from litestar.utils.predicates import is_async_callable
+
+__all__ = ("BaseLocalFileSystem", "FileSystemAdapter")
+
+
+if TYPE_CHECKING:
+    from os import stat_result
+
+    from _typeshed import OpenBinaryMode
+
+    from litestar.types import PathType
+    from litestar.types.file_types import FileInfo
+
+
+class BaseLocalFileSystem(FileSystemProtocol):
+    """Base class for a local file system."""
+
+    async def info(self, path: PathType, **kwargs: Any) -> FileInfo:
+        """Retrieve information about a given file path.
+
+        Args:
+            path: A file path.
+            **kwargs: Any additional kwargs.
+
+        Returns:
+            A dictionary of file info.
+        """
+        result = await Path(path).stat()
+        return await FileSystemAdapter.parse_stat_result(path=path, result=result)
+
+    async def open(self, file: PathType, mode: str, buffering: int = -1) -> AsyncFile[AnyStr]:  # pyright: ignore
+        """Return a file-like object from the filesystem.
+
+        Notes:
+            - The return value must be a context-manager
+
+        Args:
+            file: Path to the target file.
+            mode: Mode, similar to the built ``open``.
+            buffering: Buffer size.
+        """
+        return await open_file(file=file, mode=mode, buffering=buffering)  # type: ignore[call-overload, no-any-return]
+
+
+class FileSystemAdapter:
+    """Wrapper around a ``FileSystemProtocol``, normalising its interface."""
+
+    def __init__(self, file_system: FileSystemProtocol) -> None:
+        """Initialize an adapter from a given ``file_system``
+
+        Args:
+            file_system: A filesystem class adhering to the :class:`FileSystemProtocol <litestar.types.FileSystemProtocol>`
+        """
+        self.file_system = file_system
+
+    async def info(self, path: PathType) -> FileInfo:
+        """Proxies the call to the underlying FS Spec's ``info`` method, ensuring it's done in an async fashion and with
+        strong typing.
+
+        Args:
+            path: A file path to load the info for.
+
+        Returns:
+            A dictionary of file info.
+        """
+        try:
+            awaitable = (
+                self.file_system.info(str(path))
+                if is_async_callable(self.file_system.info)
+                else sync_to_thread(self.file_system.info, str(path))
+            )
+            return cast("FileInfo", await awaitable)
+        except FileNotFoundError as e:
+            raise e
+        except PermissionError as e:
+            raise NotAuthorizedException(f"failed to read {path} due to missing permissions") from e
+        except OSError as e:  # pragma: no cover
+            raise InternalServerException from e
+
+    async def open(
+        self,
+        file: PathType,
+        mode: OpenBinaryMode = "rb",
+        buffering: int = -1,
+    ) -> AsyncFile[bytes]:
+        """Return a file-like object from the filesystem.
+
+        Notes:
+            - The return value must function correctly in a context ``with`` block.
+
+        Args:
+            file: Path to the target file.
+            mode: Mode, similar to the built ``open``.
+            buffering: Buffer size.
+        """
+        try:
+            if is_async_callable(self.file_system.open):  # pyright: ignore
+                return cast(
+                    "AsyncFile[bytes]",
+                    await self.file_system.open(
+                        file=file,
+                        mode=mode,
+                        buffering=buffering,
+                    ),
+                )
+            return AsyncFile(await sync_to_thread(self.file_system.open, file, mode, buffering))  # type: ignore[arg-type]
+        except PermissionError as e:
+            raise NotAuthorizedException(f"failed to open {file} due to missing permissions") from e
+        except OSError as e:
+            raise InternalServerException from e
+
+    @staticmethod
+    async def parse_stat_result(path: PathType, result: stat_result) -> FileInfo:
+        """Convert a ``stat_result`` instance into a ``FileInfo``.
+
+        Args:
+            path: The file path for which the :func:`stat_result <os.stat_result>` is provided.
+            result: The :func:`stat_result <os.stat_result>` instance.
+
+        Returns:
+            A dictionary of file info.
+        """
+        file_info: FileInfo = {
+            "created": result.st_ctime,
+            "gid": result.st_gid,
+            "ino": result.st_ino,
+            "islink": await Path(path).is_symlink(),
+            "mode": result.st_mode,
+            "mtime": result.st_mtime,
+            "name": str(path),
+            "nlink": result.st_nlink,
+            "size": result.st_size,
+            "type": "directory" if S_ISDIR(result.st_mode) else "file",
+            "uid": result.st_uid,
+        }
+
+        if file_info["islink"]:
+            file_info["destination"] = str(await Path(path).readlink()).encode("utf-8")
+            try:
+                file_info["size"] = (await Path(path).stat()).st_size
+            except OSError:  # pragma: no cover
+                file_info["size"] = result.st_size
+
+        return file_info