text.py 1.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364
  1. from ..base import AsyncBase, AsyncIndirectBase
  2. from .utils import delegate_to_executor, proxy_method_directly, proxy_property_directly
  3. @delegate_to_executor(
  4. "close",
  5. "flush",
  6. "isatty",
  7. "read",
  8. "readable",
  9. "readline",
  10. "readlines",
  11. "seek",
  12. "seekable",
  13. "tell",
  14. "truncate",
  15. "write",
  16. "writable",
  17. "writelines",
  18. )
  19. @proxy_method_directly("detach", "fileno", "readable")
  20. @proxy_property_directly(
  21. "buffer",
  22. "closed",
  23. "encoding",
  24. "errors",
  25. "line_buffering",
  26. "newlines",
  27. "name",
  28. "mode",
  29. )
  30. class AsyncTextIOWrapper(AsyncBase):
  31. """The asyncio executor version of io.TextIOWrapper."""
  32. @delegate_to_executor(
  33. "close",
  34. "flush",
  35. "isatty",
  36. "read",
  37. "readable",
  38. "readline",
  39. "readlines",
  40. "seek",
  41. "seekable",
  42. "tell",
  43. "truncate",
  44. "write",
  45. "writable",
  46. "writelines",
  47. )
  48. @proxy_method_directly("detach", "fileno", "readable")
  49. @proxy_property_directly(
  50. "buffer",
  51. "closed",
  52. "encoding",
  53. "errors",
  54. "line_buffering",
  55. "newlines",
  56. "name",
  57. "mode",
  58. )
  59. class AsyncTextIndirectIOWrapper(AsyncIndirectBase):
  60. """The indirect asyncio executor version of io.TextIOWrapper."""