/opt/imh-python/lib/python3.9/site-packages/kombu/utils
NameSizeModeActions
__pycache__/-0755rm
amq_manager.py7160644editdlrm
collections.py9420644editdlrm
compat.py34440644editdlrm
debug.py20490644editdlrm
div.py9410644editdlrm
encoding.py22970644editdlrm
eventio.py101590644editdlrm
functional.py106990644editdlrm
imports.py20890644editdlrm
json.py40880644editdlrm
limits.py25510644editdlrm
objects.py20430644editdlrm
scheduling.py29270644editdlrm
text.py22000644editdlrm
time.py2720644editdlrm
url.py39090644editdlrm
uuid.py3270644editdlrm
__init__.py6980644editdlrm
Edit: /opt/imh-python/lib/python3.9/site-packages/kombu/utils/div.py (941B)
"""Div. Utilities.""" from __future__ import annotations import os import sys from .encoding import default_encode def emergency_dump_state(state, open_file=open, dump=None, stderr=None): """Dump message state to stdout or file.""" from pprint import pformat from tempfile import mkstemp stderr = sys.stderr if stderr is None else stderr if dump is None: import pickle dump = pickle.dump fd, persist = mkstemp() os.close(fd) print(f'EMERGENCY DUMP STATE TO FILE -> {persist} <-', file=stderr) fh = open_file(persist, 'w') try: try: dump(state, fh, protocol=0) except Exception as exc: print( f'Cannot pickle state: {exc!r}. Fallback to pformat.', file=stderr, ) fh.write(default_encode(pformat(state))) finally: fh.flush() fh.close() return persist