contexts.py 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151
  1. from __future__ import annotations
  2. from contextlib import contextmanager
  3. import os
  4. import sys
  5. from typing import (
  6. IO,
  7. TYPE_CHECKING,
  8. )
  9. from pandas.compat import CHAINED_WARNING_DISABLED
  10. from pandas.errors import ChainedAssignmentError
  11. from pandas.io.common import get_handle
  12. if TYPE_CHECKING:
  13. from collections.abc import Generator
  14. from pandas._typing import (
  15. BaseBuffer,
  16. CompressionOptions,
  17. FilePath,
  18. )
  19. @contextmanager
  20. def decompress_file(
  21. path: FilePath | BaseBuffer, compression: CompressionOptions
  22. ) -> Generator[IO[bytes]]:
  23. """
  24. Open a compressed file and return a file object.
  25. Parameters
  26. ----------
  27. path : str
  28. The path where the file is read from.
  29. compression : {'gzip', 'bz2', 'zip', 'xz', 'zstd', None}
  30. Name of the decompression to use
  31. Returns
  32. -------
  33. file object
  34. """
  35. with get_handle(path, "rb", compression=compression, is_text=False) as handle:
  36. yield handle.handle
  37. @contextmanager
  38. def set_timezone(tz: str) -> Generator[None]:
  39. """
  40. Context manager for temporarily setting a timezone.
  41. Parameters
  42. ----------
  43. tz : str
  44. A string representing a valid timezone.
  45. Examples
  46. --------
  47. >>> from datetime import datetime
  48. >>> from dateutil.tz import tzlocal
  49. >>> tzlocal().tzname(datetime(2021, 1, 1)) # doctest: +SKIP
  50. 'IST'
  51. >>> with set_timezone("US/Eastern"):
  52. ... tzlocal().tzname(datetime(2021, 1, 1))
  53. 'EST'
  54. """
  55. import time
  56. def setTZ(tz) -> None:
  57. if hasattr(time, "tzset"):
  58. if tz is None:
  59. try:
  60. del os.environ["TZ"]
  61. except KeyError:
  62. pass
  63. else:
  64. os.environ["TZ"] = tz
  65. # Next line allows typing checks to pass on Windows
  66. if sys.platform != "win32":
  67. time.tzset()
  68. orig_tz = os.environ.get("TZ")
  69. setTZ(tz)
  70. try:
  71. yield
  72. finally:
  73. setTZ(orig_tz)
  74. @contextmanager
  75. def with_csv_dialect(name: str, **kwargs) -> Generator[None]:
  76. """
  77. Context manager to temporarily register a CSV dialect for parsing CSV.
  78. Parameters
  79. ----------
  80. name : str
  81. The name of the dialect.
  82. kwargs : mapping
  83. The parameters for the dialect.
  84. Raises
  85. ------
  86. ValueError : the name of the dialect conflicts with a builtin one.
  87. See Also
  88. --------
  89. csv : Python's CSV library.
  90. """
  91. import csv
  92. _BUILTIN_DIALECTS = {"excel", "excel-tab", "unix"}
  93. if name in _BUILTIN_DIALECTS:
  94. raise ValueError("Cannot override builtin dialect.")
  95. csv.register_dialect(name, **kwargs)
  96. try:
  97. yield
  98. finally:
  99. csv.unregister_dialect(name)
  100. def raises_chained_assignment_error(extra_warnings=(), extra_match=()):
  101. from pandas._testing import assert_produces_warning
  102. if CHAINED_WARNING_DISABLED:
  103. if not extra_warnings:
  104. from contextlib import nullcontext
  105. return nullcontext()
  106. else:
  107. return assert_produces_warning(
  108. extra_warnings,
  109. match=extra_match,
  110. )
  111. else:
  112. warning = ChainedAssignmentError
  113. match = (
  114. "A value is being set on a copy of a DataFrame or Series "
  115. "through chained assignment"
  116. )
  117. if extra_warnings:
  118. warning = (warning, *extra_warnings) # type: ignore[assignment]
  119. return assert_produces_warning(
  120. warning,
  121. match=(match, *extra_match),
  122. )