Classes
ResourceManager
Centralized resource lifecycle management.
Ensures all resources are cleaned up regardless of how …
ResourceManager
Centralized resource lifecycle management.
Ensures all resources are cleaned up regardless of how the process exits.
Usage:
with ResourceManager() as rm:
server = rm.register_server(httpd)
observer = rm.register_observer(watcher)
# Resources automatically cleaned up on exit
Features:
- Idempotent cleanup (safe to call multiple times)
- LIFO cleanup order (like context managers)
- Timeout protection (won't hang forever)
- Thread-safe registration
- Handles all termination scenarios
Methods 5
register
Register a resource with its cleanup function.
register
def register(self, name: str, resource: Any, cleanup_fn: Callable[[Any], None]) -> Any
Register a resource with its cleanup function.
Parameters 3
name |
str |
Human-readable name for debugging |
resource |
Any |
The resource object |
cleanup_fn |
Callable[[Any], None] |
Function to call to clean up (takes resource as arg) |
Returns
The resource (for chaining)Any
—
register_server
Register HTTP server for cleanup.
register_server
def register_server(self, server: Any) -> Any
Register HTTP server for cleanup.
Parameters 1
server |
Any |
socketserver.TCPServer instance |
Returns
The serverAny
—
register_observer
Register file system observer for cleanup.
register_observer
def register_observer(self, observer: Any) -> Any
Register file system observer for cleanup.
Parameters 1
observer |
Any |
watchdog.observers.Observer instance |
Returns
The observerAny
—
register_pidfile
Register PID file for cleanup.
register_pidfile
def register_pidfile(self, pidfile_path: Path) -> Path
Register PID file for cleanup.
Parameters 1
pidfile_path |
Path |
Path object to PID file |
Returns
The pathPath
—
cleanup
Clean up all resources (idempotent).
cleanup
def cleanup(self, signum: int | None = None) -> None
Clean up all resources (idempotent).
Parameters 1
signum |
int | None |
Signal number if cleanup triggered by signal |
Internal Methods 6
__init__
Initialize resource manager.
__init__
def __init__(self) -> None
Initialize resource manager.
_signal_handler
Handle termination signals.
_signal_handler
def _signal_handler(self, signum: int, frame: FrameType | None) -> None
Handle termination signals.
Parameters 2
signum |
int |
|
frame |
FrameType | None |
_register_signal_handlers
Register signal handlers for cleanup.
_register_signal_handlers
def _register_signal_handlers(self) -> None
Register signal handlers for cleanup.
_restore_signals
Restore original signal handlers.
_restore_signals
def _restore_signals(self) -> None
Restore original signal handlers.
__enter__
Context manager entry.
__enter__
def __enter__(self) -> ResourceManager
Context manager entry.
Returns
ResourceManager
__exit__
Context manager exit - ensure cleanup runs.
__exit__
def __exit__(self, exc_type: type[BaseException] | None, *args: Any) -> None
Context manager exit - ensure cleanup runs.
Parameters 1
exc_type |
type[BaseException] | None |