1  from __future__ import annotations
   2  
   3  from collections.abc import Callable, Mapping
   4  from io import SEEK_SET, UnsupportedOperation
   5  from os import PathLike
   6  from pathlib import Path
   7  from typing import Any, BinaryIO, cast
   8  
   9  from .. import (
  10      BrokenResourceError,
  11      ClosedResourceError,
  12      EndOfStream,
  13      TypedAttributeSet,
  14      to_thread,
  15      typed_attribute,
  16  )
  17  from ..abc import ByteReceiveStream, ByteSendStream
  18  
  19  
  20  class FileStreamAttribute(TypedAttributeSet):
  21      #: the open file descriptor
  22      file: BinaryIO = typed_attribute()
  23      #: the path of the file on the file system, if available (file must be a real file)
  24      path: Path = typed_attribute()
  25      #: the file number, if available (file must be a real file or a TTY)
  26      fileno: int = typed_attribute()
  27  
  28  
  29  class _BaseFileStream:
  30      def __init__(self, file: BinaryIO):
  31          self._file = file
  32  
  33      async def aclose(self) -> None:
  34          await to_thread.run_sync(self._file.close)
  35  
  36      @property
  37      def extra_attributes(self) -> Mapping[Any, Callable[[], Any]]:
  38          attributes: dict[Any, Callable[[], Any]] = {
  39              FileStreamAttribute.file: lambda: self._file,
  40          }
  41  
  42          if hasattr(self._file, "name"):
  43              attributes[FileStreamAttribute.path] = lambda: Path(self._file.name)
  44  
  45          try:
  46              self._file.fileno()
  47          except UnsupportedOperation:
  48              pass
  49          else:
  50              attributes[FileStreamAttribute.fileno] = lambda: self._file.fileno()
  51  
  52          return attributes
  53  
  54  
  55  class FileReadStream(_BaseFileStream, ByteReceiveStream):
  56      """
  57      A byte stream that reads from a file in the file system.
  58  
  59      :param file: a file that has been opened for reading in binary mode
  60  
  61      .. versionadded:: 3.0
  62      """
  63  
  64      @classmethod
  65      async def from_path(cls, path: str | PathLike[str]) -> FileReadStream:
  66          """
  67          Create a file read stream by opening the given file.
  68  
  69          :param path: path of the file to read from
  70  
  71          """
  72          file = await to_thread.run_sync(Path(path).open, "rb")
  73          return cls(cast(BinaryIO, file))
  74  
  75      async def receive(self, max_bytes: int = 65536) -> bytes:
  76          try:
  77              data = await to_thread.run_sync(self._file.read, max_bytes)
  78          except ValueError:
  79              raise ClosedResourceError from None
  80          except OSError as exc:
  81              raise BrokenResourceError from exc
  82  
  83          if data:
  84              return data
  85          else:
  86              raise EndOfStream
  87  
  88      async def seek(self, position: int, whence: int = SEEK_SET) -> int:
  89          """
  90          Seek the file to the given position.
  91  
  92          .. seealso:: :meth:`io.IOBase.seek`
  93  
  94          .. note:: Not all file descriptors are seekable.
  95  
  96          :param position: position to seek the file to
  97          :param whence: controls how ``position`` is interpreted
  98          :return: the new absolute position
  99          :raises OSError: if the file is not seekable
 100  
 101          """
 102          return await to_thread.run_sync(self._file.seek, position, whence)
 103  
 104      async def tell(self) -> int:
 105          """
 106          Return the current stream position.
 107  
 108          .. note:: Not all file descriptors are seekable.
 109  
 110          :return: the current absolute position
 111          :raises OSError: if the file is not seekable
 112  
 113          """
 114          return await to_thread.run_sync(self._file.tell)
 115  
 116  
 117  class FileWriteStream(_BaseFileStream, ByteSendStream):
 118      """
 119      A byte stream that writes to a file in the file system.
 120  
 121      :param file: a file that has been opened for writing in binary mode
 122  
 123      .. versionadded:: 3.0
 124      """
 125  
 126      @classmethod
 127      async def from_path(
 128          cls, path: str | PathLike[str], append: bool = False
 129      ) -> FileWriteStream:
 130          """
 131          Create a file write stream by opening the given file for writing.
 132  
 133          :param path: path of the file to write to
 134          :param append: if ``True``, open the file for appending; if ``False``, any
 135              existing file at the given path will be truncated
 136  
 137          """
 138          mode = "ab" if append else "wb"
 139          file = await to_thread.run_sync(Path(path).open, mode)
 140          return cls(cast(BinaryIO, file))
 141  
 142      async def send(self, item: bytes) -> None:
 143          try:
 144              await to_thread.run_sync(self._file.write, item)
 145          except ValueError:
 146              raise ClosedResourceError from None
 147          except OSError as exc:
 148              raise BrokenResourceError from exc
 149