jmcore._notification_worker

jmcore._notification_worker

Isolated Apprise delivery worker for notifications.

Apprise HTTP plugins use process-global proxy environment variables. Keep that state in a dedicated child process so notifications cannot affect the parent JoinMarket process or any of its concurrent tasks.

Attributes

PROXY_ENVIRONMENT_KEYS = ('HTTP_PROXY', 'HTTPS_PROXY', 'http_proxy', 'https_proxy', 'ALL_PROXY', 'all_proxy', 'NO_PROXY', 'no_proxy') module-attribute

Classes

AppriseWorker

A single-process, serialized Apprise delivery worker.

Source code in jmcore/src/jmcore/_notification_worker.py

class AppriseWorker:
    """A single-process, serialized Apprise delivery worker."""

    def __init__(self, config: NotificationWorkerConfig):
        self._config = config
        self._context = mp.get_context("spawn")
        self._request_queue: Any | None = None
        self._response_queue: Any | None = None
        self._process: Any | None = None
        self._closed = False
        _active_workers.add(self)

    @property
    def closed(self) -> bool:
        """Return whether this worker has been shut down."""
        return self._closed

    def start(self) -> NotificationWorkerResult:
        """Spawn the worker and wait for its Apprise initialization result."""
        if self._closed:
            return NotificationWorkerResult(False, _WORKER_STOPPED_DIAGNOSTIC)
        try:
            self._request_queue = self._context.Queue(maxsize=1)
            self._response_queue = self._context.Queue(maxsize=1)
            self._process = self._context.Process(
                target=_notification_worker_main,
                args=(self._config, self._request_queue, self._response_queue),
                name="jm-notification-worker",
                daemon=True,
            )
            self._process.start()
            result = self._receive_response(_WORKER_START_TIMEOUT)
        except Exception:
            self.close()
            return NotificationWorkerResult(False, _GENERIC_INITIALIZATION_DIAGNOSTIC)

        if result is None:
            self.close()
            return NotificationWorkerResult(False, _INITIALIZATION_TIMEOUT_DIAGNOSTIC)
        if not result.success:
            self.close()
        return result

    def send(self, title: str, body: str, priority: str) -> NotificationWorkerResult:
        """Queue one serialized request and return its delivery result."""
        if self._closed or self._request_queue is None:
            return NotificationWorkerResult(False, _WORKER_STOPPED_DIAGNOSTIC)
        try:
            self._request_queue.put(("send", title, body, priority), timeout=1.0)
        except (OSError, queue.Full, ValueError):
            self.close()
            return NotificationWorkerResult(False, _WORKER_STOPPED_DIAGNOSTIC)

        result = self._receive_response(_WORKER_SEND_TIMEOUT)
        if result is None:
            self.close()
            return NotificationWorkerResult(False, _WORKER_STOPPED_DIAGNOSTIC)
        return result

    def close(self) -> None:
        """Stop the child without allowing shutdown to block indefinitely."""
        if self._closed:
            return
        self._closed = True
        process = self._process
        request_queue = self._request_queue

        if process is not None and process.is_alive():
            if request_queue is not None:
                with suppress(OSError, queue.Full, ValueError):
                    request_queue.put_nowait(("close",))
            process.join(_WORKER_JOIN_TIMEOUT)
            if process.is_alive():
                process.terminate()
                process.join(_WORKER_JOIN_TIMEOUT)

        for worker_queue in (self._request_queue, self._response_queue):
            if worker_queue is not None:
                with suppress(OSError, ValueError):
                    worker_queue.close()
        self._process = None
        self._request_queue = None
        self._response_queue = None
        _active_workers.discard(self)

    def _receive_response(self, timeout: float) -> NotificationWorkerResult | None:
        """Wait in short intervals so close or a crashed child is noticed promptly."""
        if self._response_queue is None:
            return None
        deadline = time.monotonic() + timeout
        while True:
            if self._closed:
                return None
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return None
            try:
                response = self._response_queue.get(timeout=min(remaining, 0.1))
                if isinstance(response, NotificationWorkerResult):
                    return response
                return NotificationWorkerResult(False, _WORKER_STOPPED_DIAGNOSTIC)
            except queue.Empty:
                if self._process is None or not self._process.is_alive():
                    return None
            except (EOFError, OSError, ValueError):
                return None

Attributes

closed: bool property

Return whether this worker has been shut down.

Methods:

__init__(config: NotificationWorkerConfig)

Source code in jmcore/src/jmcore/_notification_worker.py

def __init__(self, config: NotificationWorkerConfig):
    self._config = config
    self._context = mp.get_context("spawn")
    self._request_queue: Any | None = None
    self._response_queue: Any | None = None
    self._process: Any | None = None
    self._closed = False
    _active_workers.add(self)

close() -> None

Stop the child without allowing shutdown to block indefinitely.

Source code in jmcore/src/jmcore/_notification_worker.py

def close(self) -> None:
    """Stop the child without allowing shutdown to block indefinitely."""
    if self._closed:
        return
    self._closed = True
    process = self._process
    request_queue = self._request_queue

    if process is not None and process.is_alive():
        if request_queue is not None:
            with suppress(OSError, queue.Full, ValueError):
                request_queue.put_nowait(("close",))
        process.join(_WORKER_JOIN_TIMEOUT)
        if process.is_alive():
            process.terminate()
            process.join(_WORKER_JOIN_TIMEOUT)

    for worker_queue in (self._request_queue, self._response_queue):
        if worker_queue is not None:
            with suppress(OSError, ValueError):
                worker_queue.close()
    self._process = None
    self._request_queue = None
    self._response_queue = None
    _active_workers.discard(self)

send(title: str, body: str, priority: str) -> NotificationWorkerResult

Queue one serialized request and return its delivery result.

Source code in jmcore/src/jmcore/_notification_worker.py

def send(self, title: str, body: str, priority: str) -> NotificationWorkerResult:
    """Queue one serialized request and return its delivery result."""
    if self._closed or self._request_queue is None:
        return NotificationWorkerResult(False, _WORKER_STOPPED_DIAGNOSTIC)
    try:
        self._request_queue.put(("send", title, body, priority), timeout=1.0)
    except (OSError, queue.Full, ValueError):
        self.close()
        return NotificationWorkerResult(False, _WORKER_STOPPED_DIAGNOSTIC)

    result = self._receive_response(_WORKER_SEND_TIMEOUT)
    if result is None:
        self.close()
        return NotificationWorkerResult(False, _WORKER_STOPPED_DIAGNOSTIC)
    return result

start() -> NotificationWorkerResult

Spawn the worker and wait for its Apprise initialization result.

Source code in jmcore/src/jmcore/_notification_worker.py

def start(self) -> NotificationWorkerResult:
    """Spawn the worker and wait for its Apprise initialization result."""
    if self._closed:
        return NotificationWorkerResult(False, _WORKER_STOPPED_DIAGNOSTIC)
    try:
        self._request_queue = self._context.Queue(maxsize=1)
        self._response_queue = self._context.Queue(maxsize=1)
        self._process = self._context.Process(
            target=_notification_worker_main,
            args=(self._config, self._request_queue, self._response_queue),
            name="jm-notification-worker",
            daemon=True,
        )
        self._process.start()
        result = self._receive_response(_WORKER_START_TIMEOUT)
    except Exception:
        self.close()
        return NotificationWorkerResult(False, _GENERIC_INITIALIZATION_DIAGNOSTIC)

    if result is None:
        self.close()
        return NotificationWorkerResult(False, _INITIALIZATION_TIMEOUT_DIAGNOSTIC)
    if not result.success:
        self.close()
    return result

NotificationWorker

Bases: Protocol

Synchronous interface used by :class:jmcore.notifications.Notifier.

Source code in jmcore/src/jmcore/_notification_worker.py

class NotificationWorker(Protocol):
    """Synchronous interface used by :class:`jmcore.notifications.Notifier`."""

    @property
    def closed(self) -> bool:
        """Whether the worker can no longer accept requests."""
        ...

    def start(self) -> NotificationWorkerResult:
        """Start the worker and initialize its Apprise services."""
        ...

    def send(self, title: str, body: str, priority: str) -> NotificationWorkerResult:
        """Deliver one notification."""
        ...

    def close(self) -> None:
        """Stop the worker and release its resources."""
        ...

Attributes

closed: bool property

Whether the worker can no longer accept requests.

Methods:

close() -> None

Stop the worker and release its resources.

Source code in jmcore/src/jmcore/_notification_worker.py

def close(self) -> None:
    """Stop the worker and release its resources."""
    ...

send(title: str, body: str, priority: str) -> NotificationWorkerResult

Deliver one notification.

Source code in jmcore/src/jmcore/_notification_worker.py

def send(self, title: str, body: str, priority: str) -> NotificationWorkerResult:
    """Deliver one notification."""
    ...

start() -> NotificationWorkerResult

Start the worker and initialize its Apprise services.

Source code in jmcore/src/jmcore/_notification_worker.py

def start(self) -> NotificationWorkerResult:
    """Start the worker and initialize its Apprise services."""
    ...

NotificationWorkerConfig dataclass

Configuration transferred to the isolated notification worker.

Source code in jmcore/src/jmcore/_notification_worker.py

@dataclass(frozen=True)
class NotificationWorkerConfig:
    """Configuration transferred to the isolated notification worker."""

    urls: tuple[str, ...]
    use_tor: bool
    tor_socks_host: str
    tor_socks_port: int
    stream_isolation: bool

Attributes

stream_isolation: bool instance-attribute

tor_socks_host: str instance-attribute

tor_socks_port: int instance-attribute

urls: tuple[str, ...] instance-attribute

use_tor: bool instance-attribute

NotificationWorkerResult dataclass

A bounded delivery result that contains no notification content.

Source code in jmcore/src/jmcore/_notification_worker.py

@dataclass(frozen=True)
class NotificationWorkerResult:
    """A bounded delivery result that contains no notification content."""

    success: bool
    diagnostic: str | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "diagnostic", _sanitize_worker_diagnostic(self.diagnostic))

Attributes

diagnostic: str | None = None class-attribute instance-attribute

success: bool instance-attribute

Methods:

__post_init__() -> None

Source code in jmcore/src/jmcore/_notification_worker.py

def __post_init__(self) -> None:
    object.__setattr__(self, "diagnostic", _sanitize_worker_diagnostic(self.diagnostic))

Functions: