Merge pull request #1 from caliangroup/calian-updates

Calian updates to sdnotify

.gitignore

@@ -25,3 +25,7 @@ var/
 *.egg-info/
 .installed.cfg
 *.egg
+
+# Python
+.coverage
+coverage.xml

.vscode/settings.json

@@ -0,0 +1,15 @@
+{
+    "python.analysis.autoImportCompletions": true,
+    "python.testing.pytestArgs": [
+        "-v",
+        "--cov=aiosdnotify",
+        "--cov-report=xml",
+        "--cov-report=term",
+        "tests"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true,
+    "mypy-type-checker.args": [
+        "--strict"
+    ]
+}

LICENSE.txt

@@ -1,6 +1,6 @@
 The MIT License (MIT)
 
-Copyright (c) 2016 Brett Bethke
+Copyright 2024 Calian Ltd. All rights reserved.
 
 Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 

README.md

@@ -1,6 +1,7 @@
 # systemd Service Notification
 
-This is Garrett's Asyncio version of the systemd sd_notify package written by bb4242 (Brett Bethke),
+This is a Calian fork of Garrett's Asyncio version (https://github.com/seatsnob/sdnotify-asyncio)
+of the systemd sd_notify package written by bb4242 (Brett Bethke),
 available at https://github.com/bb4242/sdnotify
 
 This is a pure Python implementation of the
@@ -9,11 +10,7 @@ This is a pure Python implementation of the
 protocol. This protocol can be used to inform `systemd` about service start-up
 completion, watchdog events, and other service status changes. Thus, this
 package can be used to write system services in Python that play nicely with
-`systemd`. `sdnotify` is compatible with both Python 2 and Python 3.
-
-Just so we're clear, I figured out _why_ nobody made this. Sdnotify uses UDP under the hood.
-There isn't an asyncio implementation of UDP. So I mostly just wound up writing shitty wrappers
-for asyncio around multithreading.
+`systemd`. `async-sdnotify` is compatible with Python 3.
 
 Normally the `SystemdNotifier.notify` method silently ignores exceptions (for example, if the
 systemd notification socket is not available) to allow applications to
@@ -23,7 +20,7 @@ aid in debugging.
 
 # Installation
 
-`pip install sdnotify-asyncio`
+`pip install async-sdnotify`
 
 # Example Usage
 

aiosdnotify/__init__.py

@@ -1,10 +1,20 @@
+#
+# Copyright 2024 Calian Ltd. All rights reserved.
+#
+# Original authors:
+#   - Copyright 2023 Seat Snob (garrett@seatsnob.com)
+#   - Copyright 2016 Brett Bethke (bbethke@gmail.com)
+#
+
 import socket
 import asyncio
 import os
 import logging
+from types import TracebackType
+from typing import Any, Optional, Self, Type
 
 
-__version__ = "1.0.0"
+__version__ = "0.1.0"
 
 logger = logging.getLogger(__name__)
 
@@ -13,7 +23,7 @@ class SystemdNotifier:
     """This class holds a connection to the systemd notification socket
     and can be used to send messages to systemd using its notify method."""
 
-    def __init__(self, debug=False, warn_limit=3, watchdog=True):
+    def __init__(self, debug: bool=False, warn_limit: int=3, watchdog: bool=True) -> None:
         """Instantiate a new notifier object. This will initiate a connection
         to the systemd notification socket.
 
@@ -36,29 +46,31 @@ def __init__(self, debug=False, warn_limit=3, watchdog=True):
         self._warnings = 0
         self._transport: asyncio.DatagramTransport | None = None
         self._protocol: asyncio.DatagramProtocol | None = None
-        self._watchdog_task: asyncio.Task | None = None
+        self._watchdog_task: asyncio.Task[None] | None = None
         self._ready_event = asyncio.Event()
 
-    async def connect(self):
+    async def connect(self) -> None:
         try:
-            notify_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
             addr = os.getenv('NOTIFY_SOCKET')
+            if addr is None:
+                raise EnvironmentError('NOTIFY_SOCKET not set')
+            notify_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
             if addr[0] == '@':
                 addr = '\0' + addr[1:]
             notify_socket.connect(addr)
             self._transport, self._protocol = await asyncio.get_running_loop().create_datagram_endpoint(
                 asyncio.DatagramProtocol, sock=notify_socket)
-        except Exception as e:
+        except Exception:
             if self.debug:
                 raise
             elif self.warn_limit:
                 logger.warning('SystemdNotifier failed to connect', exc_info=True)
 
-    def disconnect(self):
+    def disconnect(self) -> None:
         if self._transport is not None and not self._transport.is_closing():
             self._transport.close()
 
-    def _notify(self, state_str: str):
+    def _notify(self, state_str: str) -> None:
         """Send a notification to systemd. state is a string; see
         the man page of sd_notify (http://www.freedesktop.org/software/systemd/man/sd_notify.html)
         for a description of the allowable values.
@@ -69,6 +81,8 @@ def _notify(self, state_str: str):
         cause this method to raise any exceptions generated to the caller, to
         aid in debugging."""
         try:
+            if self._transport is None:
+                raise RuntimeError('SystemdNotifier is not connected')
             self._transport.sendto(state_str.encode())
             self._warnings = 0
         except Exception:
@@ -78,35 +92,40 @@ def _notify(self, state_str: str):
                 logger.warning('SystemdNotifier failed to notify', exc_info=True)
                 self._warnings += 1
 
-    def ready(self):
+    def ready(self) -> None:
         self._notify("READY=1")
         self._ready_event.set()
 
-    def status(self, status: str):
+    def status(self, status: str) -> None:
         self._notify(f"STATUS={status}")
 
-    def start_watchdog(self, interval: float | None = None):
+    def start_watchdog(self, interval: float | None = None) -> None:
         if interval is None:
-            interval_microseconds = float(os.getenv('WATCHDOG_USEC'))
+            interval_microseconds = os.getenv('WATCHDOG_USEC')
             if interval_microseconds is None:
-                raise EnvironmentError('Unable to determine watchdog interval from ENV, and none was specified')
-            interval = interval_microseconds / 1_000_000 / 2  # We try to notify at half the env specified interval
+                message = 'Unable to determine watchdog interval from ENV, and none was specified'
+                if self.debug:
+                    raise EnvironmentError(message)
+                else:
+                    logger.warning(message)
+                    return
+            interval = float(interval_microseconds) / 1_000_000 / 2  # We try to notify at half the env specified interval
         self._watchdog_task = asyncio.create_task(
             self._watchdog(interval))
 
-    async def _watchdog(self, interval: float):
+    async def _watchdog(self, interval: float) -> None:
         await self._ready_event.wait()
         while True:
             self._notify('WATCHDOG=1')
             await asyncio.sleep(interval)
 
-    async def __aenter__(self):
+    async def __aenter__(self) -> Self:
         await self.connect()
         if self.watchdog:
             self.start_watchdog()
         return self
 
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
+    async def __aexit__(self, exc_type: Type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None:
         if (
                 self._watchdog_task
                 and not self._watchdog_task.done()