Module

server.resource_manager

Resource lifecycle management for Bengal dev server.

Provides centralized cleanup handling for all termination scenarios:

  • Normal exit (context manager)
  • Ctrl+C (KeyboardInterrupt + signal handler)
  • kill/SIGTERM (signal handler)
  • Parent death (atexit handler)
  • Exceptions (context manager exit)

Classes

ResourceManager
Centralized resource lifecycle management. Ensures all resources are cleaned up regardless of how …
11

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.
3 Any
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

Any

The resource (for chaining)

register_server
Register HTTP server for cleanup.
1 Any
def register_server(self, server: Any) -> Any

Register HTTP server for cleanup.

Parameters 1
server Any

socketserver.TCPServer instance

Returns

Any

The server

register_observer
Register file system observer for cleanup.
1 Any
def register_observer(self, observer: Any) -> Any

Register file system observer for cleanup.

Parameters 1
observer Any

watchdog.observers.Observer instance

Returns

Any

The observer

register_pidfile
Register PID file for cleanup.
1 Path
def register_pidfile(self, pidfile_path: Path) -> Path

Register PID file for cleanup.

Parameters 1
pidfile_path Path

Path object to PID file

Returns

Path

The path

cleanup
Clean up all resources (idempotent).
1 None
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.
0 None
def __init__(self) -> None

Initialize resource manager.

_signal_handler
Handle termination signals.
2 None
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.
0 None
def _register_signal_handlers(self) -> None

Register signal handlers for cleanup.

_restore_signals
Restore original signal handlers.
0 None
def _restore_signals(self) -> None

Restore original signal handlers.

__enter__
Context manager entry.
0 ResourceManager
def __enter__(self) -> ResourceManager

Context manager entry.

Returns

ResourceManager

__exit__
Context manager exit - ensure cleanup runs.
1 None
def __exit__(self, exc_type: type[BaseException] | None, *args: Any) -> None

Context manager exit - ensure cleanup runs.

Parameters 1
exc_type type[BaseException] | None