modules: add typing

All code changes should not impact current users.

Force Sphinx to use the docstrings from the actual extension modules, otherwise it would only parse the new stub files.

Author: viccie30

docs/conf.py

@@ -20,6 +20,9 @@
 # documentation root, use os.path.abspath to make it absolute, like shown here.
 #sys.path.insert(0, os.path.abspath('.'))
 
+# Ignore stub files when generating documentation for extension modules.
+os.environ['SPHINX_AUTODOC_IGNORE_NATIVE_MODULE_TYPE_STUBS'] = '1'
+
 # -- General configuration -----------------------------------------------------
 
 # If your documentation needs a minimal Sphinx version, state it here.

pyproject.toml

@@ -43,6 +43,7 @@ build = [
     "meson>=1.8.2",
     "twine>=4.0.2",
 ]
+typing = ["mypy"]
 
 [build-system]
 requires = ["meson-python"]

src/systemd/_daemon.pyi

@@ -0,0 +1,41 @@
+import socket
+from typing import Final, SupportsIndex
+
+from typing_extensions import Unpack
+from _typeshed import StrOrBytesPath, SupportsLenAndGetItem
+
+__version__: Final[str]
+
+LISTEN_FDS_START: Final[int]
+
+def booted() -> bool: ...
+def notify(status: str,
+           unset_environment: bool = False,
+           pid: SupportsIndex = 0,
+           fds: SupportsLenAndGetItem[SupportsIndex] = []) -> bool: ...
+def _listen_fds(unset_environment: bool = True) -> int: ...
+def _listen_fds_with_names(unset_environment: bool = True) -> tuple[int, Unpack[tuple[str, ...]]]: ...
+def _is_fifo(fd: SupportsIndex, path: StrOrBytesPath | None = None, /) -> bool: ...
+def _is_mq(fd: SupportsIndex, path: StrOrBytesPath | None = None, /) -> bool: ...
+def _is_socket(fd: SupportsIndex,
+               family: SupportsIndex = socket.AF_UNSPEC,
+               type: SupportsIndex = 0,
+               listening: SupportsIndex = -1,
+               /) -> bool: ...
+def _is_socket_inet(fd: SupportsIndex,
+                    family: SupportsIndex = socket.AF_UNSPEC,
+                    type: SupportsIndex = 0,
+                    listening: SupportsIndex = -1,
+                    port: SupportsIndex = 0,
+                    /) -> bool: ...
+def _is_socket_sockaddr(fd: SupportsIndex,
+                        address: str,
+                        type: SupportsIndex = 0,
+                        flowinfo: SupportsIndex = 0,
+                        listening: SupportsIndex = -1,
+                        /) -> bool: ...
+def _is_socket_unix(fd: SupportsIndex,
+                    type: SupportsIndex = 0,
+                    listening: SupportsIndex = -1,
+                    path: StrOrBytesPath | None = None,
+                    /) -> bool: ...

src/systemd/_journal.pyi

@@ -0,0 +1,8 @@
+from typing import Final, SupportsIndex
+
+__version__: Final[str]
+
+def sendv(*args: str | bytes) -> None: ...
+def stream_fd(identifier: str,
+              priority: SupportsIndex,
+              level_prefix: SupportsIndex) -> int: ...

src/systemd/_reader.pyi

@@ -0,0 +1,120 @@
+import datetime
+import enum
+from types import TracebackType
+from typing import Any, Final, SupportsIndex, TypeVar, final, overload
+import uuid
+
+from typing_extensions import Self
+from _typeshed import structseq, SupportsLenAndGetItem, StrOrBytesPath
+
+_Timestamp = TypeVar("_Timestamp", int, datetime.timedelta)
+_Bootid = TypeVar("_Bootid", bytes, uuid.UUID)
+
+__version__: Final[str]
+
+class OpenFlag(enum.IntFlag):
+    LOCAL_ONLY = ...
+    RUNTIME_ONLY = ...
+    SYSTEM = ...
+    CURRENT_USER = ...
+    OS_ROOT = ...
+    ALL_NAMESPACES = ...
+    INCLUDE_DEFAULT_NAMESPACE = ...
+    TAKE_DIRECTORY_FD = ...
+    ASSUME_IMMUTABLE = ...
+    SYSTEM_ONLY = ...
+
+LOCAL_ONLY = OpenFlag.LOCAL_ONLY
+RUNTIME_ONLY = OpenFlag.RUNTIME_ONLY
+SYSTEM = OpenFlag.SYSTEM
+CURRENT_USER = OpenFlag.CURRENT_USER
+OS_ROOT = OpenFlag.OS_ROOT
+ALL_NAMESPACES = OpenFlag.ALL_NAMESPACES
+INCLUDE_DEFAULT_NAMESPACE = OpenFlag.INCLUDE_DEFAULT_NAMESPACE
+TAKE_DIRECTORY_FD = OpenFlag.TAKE_DIRECTORY_FD
+ASSUME_IMMUTABLE = OpenFlag.ASSUME_IMMUTABLE
+SYSTEM_ONLY = OpenFlag.SYSTEM_ONLY
+
+class WakeupEventType(enum.IntEnum):
+    NOP = ...
+    APPEND = ...
+    INVALIDATE = ...
+
+NOP = WakeupEventType.NOP
+APPEND = WakeupEventType.APPEND
+INVALIDATE = WakeupEventType.INVALIDATE
+
+def _get_catalog(id: str, /) -> str: ...
+
+class _Reader:
+    @overload
+    def __init__(self,
+                 flags: SupportsIndex = LOCAL_ONLY,
+                 *,
+                 files: SupportsLenAndGetItem[StrOrBytesPath | SupportsIndex]) -> None: ...
+    @overload
+    def __init__(self,
+                 flags: SupportsIndex = LOCAL_ONLY,
+                 *,
+                 namespace: StrOrBytesPath) -> None: ...
+    @overload
+    def __init__(self,
+                 flags: SupportsIndex = LOCAL_ONLY,
+                 path: StrOrBytesPath | SupportsIndex | None = None,
+                 files: None = None,
+                 namespace: None = None) -> None: ...
+    def fileno(self) -> int: ...
+    def reliable_fd(self) -> bool: ...
+    def get_events(self) -> int: ...
+    def get_timeout(self) -> int | None: ...
+    def get_timeout_ms(self) -> int: ...
+    def close(self) -> None: ...
+    def get_usage(self) -> int: ...
+    def __enter__(self) -> Self: ...
+    def __exit__(self,
+                 exc_type: type[BaseException] | None,
+                 exc_value: BaseException | None,
+                 traceback: TracebackType | None,
+                 /) -> None: ...
+    def _next(self, skip: SupportsIndex = 1, /) -> bool: ...
+    def _previous(self, skip: SupportsIndex = 1, /) -> bool: ...
+    def _get(self, field: str, /) -> bytes: ...
+    def _get_all(self) -> dict[str, bytes | list[bytes]]: ...
+    def _get_realtime(self) -> int: ...
+    def _get_monotonic(self) -> Monotonic[int, bytes]: ...
+    def add_match(self, match: str | bytes, /) -> None: ...
+    def add_disjunction(self) -> None: ...
+    def add_conjunction(self) -> None: ...
+    def flush_matches(self) -> None: ...
+    def seek_head(self) -> None: ...
+    def seek_tail(self) -> None: ...
+    def seek_realtime(self, timestamp: SupportsIndex, /) -> None: ...
+    def seek_monotonic(self,
+                       timestamp: SupportsIndex,
+                       bootid: str | None = None,
+                       /) -> None: ...
+    def _get_start(self) -> int: ...
+    def _get_end(self) -> int: ...
+    def process(self) -> WakeupEventType: ...
+    def wait(self, timeout: SupportsIndex = -1, /) -> WakeupEventType: ...
+    def seek_cursor(self, cursor: str, /) -> None: ...
+    def _get_cursor(self) -> str: ...
+    def test_cursor(self, cursor: str, /) -> bool: ...
+    def query_unique(self, query: str, /) -> set[bytes]: ...
+    def enumerate_fields(self) -> set[str]: ...
+    def has_runtime_files(self) -> bool: ...
+    def has_persistent_files(self) -> bool: ...
+    def get_catalog(self) -> str: ...
+    @property
+    def data_threshold(self) -> int: ...
+    @data_threshold.setter
+    def data_threshold(self, value: int) -> None: ...
+    @property
+    def closed(self) -> bool: ...
+
+@final
+class Monotonic(structseq[Any], tuple[_Timestamp, _Bootid]):
+    @property
+    def timestamp(self) -> _Timestamp: ...
+    @property
+    def bootid(self) -> _Bootid: ...

src/systemd/daemon.py

@@ -1,6 +1,9 @@
 # SPDX-License-Identifier: LGPL-2.1-or-later
+from __future__ import annotations
 
 from socket import AF_UNSPEC as _AF_UNSPEC
+import socket as _socket
+import typing as _typing
 
 from ._daemon import (__version__,
                       booted,
@@ -15,25 +18,39 @@
                       _is_mq,
                       LISTEN_FDS_START)
 
-def _convert_fileobj(fileobj):
-    try:
-        return fileobj.fileno()
-    except AttributeError:
+if _typing.TYPE_CHECKING:
+    from _typeshed import FileDescriptorLike, StrOrBytesPath
+
+def _convert_fileobj(fileobj: FileDescriptorLike) -> int:
+    if isinstance(fileobj, int):
         return fileobj
+    return fileobj.fileno()
 
-def is_fifo(fileobj, path=None):
+def is_fifo(fileobj: FileDescriptorLike,
+            path: StrOrBytesPath | None = None) -> bool:
     fd = _convert_fileobj(fileobj)
     return _is_fifo(fd, path)
 
-def is_socket(fileobj, family=_AF_UNSPEC, type=0, listening=-1):
+def is_socket(fileobj: FileDescriptorLike,
+              family: _socket.AddressFamily = _AF_UNSPEC,
+              type: _socket.SocketKind | None = None,
+              listening: int = -1) -> bool:
     fd = _convert_fileobj(fileobj)
-    return _is_socket(fd, family, type, listening)
+    return _is_socket(fd, family, type or 0, listening)
 
-def is_socket_inet(fileobj, family=_AF_UNSPEC, type=0, listening=-1, port=0):
+def is_socket_inet(fileobj: FileDescriptorLike,
+                   family: _socket.AddressFamily = _AF_UNSPEC,
+                   type: _socket.SocketKind | None = None,
+                   listening: int = -1,
+                   port: int = 0) -> bool:
     fd = _convert_fileobj(fileobj)
-    return _is_socket_inet(fd, family, type, listening, port)
+    return _is_socket_inet(fd, family, type or 0, listening, port)
 
-def is_socket_sockaddr(fileobj, address, type=0, flowinfo=0, listening=-1):
+def is_socket_sockaddr(fileobj: FileDescriptorLike,
+                       address: str,
+                       type: _socket.SocketKind | None = None,
+                       flowinfo: int = 0,
+                       listening: int = -1) -> bool:
     """Check socket type, address and/or port, flowinfo, listening state.
 
     Wraps sd_is_socket_inet_sockaddr(3).
@@ -45,17 +62,20 @@ def is_socket_sockaddr(fileobj, address, type=0, flowinfo=0, listening=-1):
     Constants for `family` are defined in the socket module.
     """
     fd = _convert_fileobj(fileobj)
-    return _is_socket_sockaddr(fd, address, type, flowinfo, listening)
+    return _is_socket_sockaddr(fd, address, type or 0, flowinfo, listening)
 
-def is_socket_unix(fileobj, type=0, listening=-1, path=None):
+def is_socket_unix(fileobj: FileDescriptorLike,
+                   type: _socket.SocketKind | None = None,
+                   listening: int = -1,
+                   path: StrOrBytesPath | None = None) -> bool:
     fd = _convert_fileobj(fileobj)
-    return _is_socket_unix(fd, type, listening, path)
+    return _is_socket_unix(fd, type or 0, listening, path)
 
-def is_mq(fileobj, path=None):
+def is_mq(fileobj: FileDescriptorLike, path: StrOrBytesPath | None = None) -> bool:
     fd = _convert_fileobj(fileobj)
     return _is_mq(fd, path)
 
-def listen_fds(unset_environment=True):
+def listen_fds(unset_environment: bool = True) -> list[int]:
     """Return a list of socket activated descriptors
 
     Example::
@@ -73,7 +93,7 @@ def listen_fds(unset_environment=True):
     num = _listen_fds(unset_environment)
     return list(range(LISTEN_FDS_START, LISTEN_FDS_START + num))
 
-def listen_fds_with_names(unset_environment=True):
+def listen_fds_with_names(unset_environment: bool = True) -> dict[int, str]:
     """Return a dictionary of socket activated descriptors as {fd: name}
 
     Example::
@@ -88,8 +108,5 @@ def listen_fds_with_names(unset_environment=True):
       Execing python3 (...)
       [3]
     """
-    composite = _listen_fds_with_names(unset_environment)
-    retval = {}
-    for i in range(0, composite[0]):
-        retval[i+LISTEN_FDS_START] = composite[1+i]
-    return retval
+    _, *names = _listen_fds_with_names(unset_environment)
+    return {i: name for i, name in enumerate(names, LISTEN_FDS_START)}