summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/file_system.py
diff options
context:
space:
mode:
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/file_system.py')
-rw-r--r--venv/lib/python3.11/site-packages/litestar/file_system.py154
1 files changed, 0 insertions, 154 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
deleted file mode 100644
index fcb77c7..0000000
--- a/venv/lib/python3.11/site-packages/litestar/file_system.py
+++ /dev/null
@@ -1,154 +0,0 @@
-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