common.py 1.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950
  1. from contextlib import contextmanager
  2. import pathlib
  3. import tempfile
  4. from typing import Generator
  5. import pytest
  6. from pandas.io.pytables import HDFStore
  7. tables = pytest.importorskip("tables")
  8. # set these parameters so we don't have file sharing
  9. tables.parameters.MAX_NUMEXPR_THREADS = 1
  10. tables.parameters.MAX_BLOSC_THREADS = 1
  11. tables.parameters.MAX_THREADS = 1
  12. def safe_close(store):
  13. try:
  14. if store is not None:
  15. store.close()
  16. except OSError:
  17. pass
  18. # contextmanager to ensure the file cleanup
  19. @contextmanager
  20. def ensure_clean_store(
  21. path, mode="a", complevel=None, complib=None, fletcher32=False
  22. ) -> Generator[HDFStore, None, None]:
  23. with tempfile.TemporaryDirectory() as tmpdirname:
  24. tmp_path = pathlib.Path(tmpdirname, path)
  25. with HDFStore(
  26. tmp_path,
  27. mode=mode,
  28. complevel=complevel,
  29. complib=complib,
  30. fletcher32=fletcher32,
  31. ) as store:
  32. yield store
  33. def _maybe_remove(store, key):
  34. """
  35. For tests using tables, try removing the table to be sure there is
  36. no content from previous tests using the same table name.
  37. """
  38. try:
  39. store.remove(key)
  40. except (ValueError, KeyError):
  41. pass