python-zeroconf · bdraco · Aug 29, 2026 · Aug 29, 2026
diff --git a/examples/async_browser.py b/examples/async_browser.py
@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+
+"""Browse for HTTP services with the asyncio API and print each event."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import contextlib
+import logging
+
+from zeroconf import IPVersion, ServiceStateChange, Zeroconf
+from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf
+
+_background_tasks: set[asyncio.Task] = set()
+
+
+def on_service_state_change(
+    zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange
+) -> None:
+    print(f"{state_change.name}: {name}")
+    if state_change is ServiceStateChange.Added:
+        task = asyncio.create_task(show_service_info(zeroconf, service_type, name))
+        _background_tasks.add(task)
+        task.add_done_callback(_background_tasks.discard)
+
+
+async def show_service_info(zeroconf: Zeroconf, service_type: str, name: str) -> None:
+    info = AsyncServiceInfo(service_type, name)
+    if await info.async_request(zeroconf, 3000):
+        print(f"  addresses: {', '.join(info.parsed_scoped_addresses())}")
+        print(f"  server: {info.server} port: {info.port}")
+        print(f"  properties: {info.decoded_properties}")
+
+
+async def main(ip_version: IPVersion) -> None:
+    aiozc = AsyncZeroconf(ip_version=ip_version)
+    browser = AsyncServiceBrowser(aiozc.zeroconf, "_http._tcp.local.", handlers=[on_service_state_change])
+    print("browsing for _http._tcp.local., press ctrl-c to exit")
+    try:
+        await asyncio.Event().wait()
+    finally:
+        await browser.async_cancel()
+        await aiozc.async_close()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--debug", action="store_true", help="enable debug logging")
+    parser.add_argument("--v6-only", action="store_true", help="use IPv6 only")
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
+    ip_version = IPVersion.V6Only if args.v6_only else IPVersion.All
+
+    with contextlib.suppress(KeyboardInterrupt):
+        asyncio.run(main(ip_version))
diff --git a/examples/async_registration.py b/examples/async_registration.py
@@ -0,0 +1,60 @@
+#!/usr/bin/env python
+
+"""Register a batch of demo HTTP services with the asyncio API."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import contextlib
+import logging
+import socket
+
+from zeroconf import IPVersion, ServiceInfo
+from zeroconf.asyncio import AsyncZeroconf
+
+SERVICE_COUNT = 250
+
+
+def build_services() -> list[ServiceInfo]:
+    return [
+        ServiceInfo(
+            "_http._tcp.local.",
+            f"Demo Web Service {i}._http._tcp.local.",
+            addresses=[socket.inet_aton("127.0.0.1")],
+            port=8080 + i,
+            properties={"path": "/"},
+            server=f"demo-host-{i}.local.",
+        )
+        for i in range(SERVICE_COUNT)
+    ]
+
+
+async def main(ip_version: IPVersion) -> None:
+    aiozc = AsyncZeroconf(ip_version=ip_version)
+    services = build_services()
+    print(f"registering {len(services)} demo services, press ctrl-c to exit")
+    background_tasks = await asyncio.gather(*(aiozc.async_register_service(service) for service in services))
+    await asyncio.gather(*background_tasks)
+    try:
+        await asyncio.Event().wait()
+    finally:
+        print("unregistering")
+        unregister_tasks = await asyncio.gather(
+            *(aiozc.async_unregister_service(service) for service in services)
+        )
+        await asyncio.gather(*unregister_tasks)
+        await aiozc.async_close()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--debug", action="store_true", help="enable debug logging")
+    parser.add_argument("--v6-only", action="store_true", help="use IPv6 only")
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
+    ip_version = IPVersion.V6Only if args.v6_only else IPVersion.All
+
+    with contextlib.suppress(KeyboardInterrupt):
+        asyncio.run(main(ip_version))
diff --git a/src/zeroconf/_cache.py b/src/zeroconf/_cache.py
@@ -210,6 +210,10 @@ def async_expire(self, now: _float) -> list[DNSRecord]:
         return expired
 
     def async_get_unique(self, entry: _UniqueRecordsType) -> DNSRecord | None:
+        """Look up the cached copy of a unique record, or None.
+
+        Event loop only; not threadsafe.
+        """
         store = self.cache.get(entry.key)
         if store is None:
             return None
@@ -232,9 +236,11 @@ def async_all_by_details(self, name: _str, type_: _int, class_: _int) -> list[DN
         return matches
 
     def async_entries_with_name(self, name: str) -> list[DNSRecord]:
+        """All cached records for a name; event loop only, not threadsafe."""
         return self.entries_with_name(name)
 
     def async_entries_with_server(self, name: str) -> list[DNSRecord]:
+        """All cached records for a server name; event loop only, not threadsafe."""
         return self.entries_with_server(name)
 
     # The below functions are threadsafe and do not need to be run in the
@@ -279,6 +285,7 @@ def get_all_by_details(self, name: str, type_: _int, class_: _int) -> list[DNSRe
         return [entry for entry in list(records.values()) if type_ == entry.type and class_ == entry.class_]
 
     def entries_with_server(self, server: str) -> list[DNSRecord]:
+        """All cached records whose server field matches."""
         if entries := self.service_cache.get(server.lower()):
             return list(entries.values())
         return []

diff --git a/src/zeroconf/_core.py b/src/zeroconf/_core.py
@@ -280,6 +280,7 @@ def listeners(self) -> set[RecordUpdateListener]:
         return self.record_manager.listeners
 
     async def async_wait(self, timeout: float) -> None:
+        """Suspend the calling task for timeout milliseconds, or less when notified."""
         loop = self.loop
         assert loop is not None
         await wait_for_future_set_or_timeout(loop, self._notify_futures, timeout)
@@ -302,6 +303,13 @@ def get_service_info(
         timeout: int = 3000,
         question_type: DNSQuestionType | None = None,
     ) -> ServiceInfo | None:
+        """Look up details for a service on the network.
+
+        Queries for the given fully qualified type and name, waiting up to
+        timeout milliseconds, and returns a populated ServiceInfo or None
+        when nothing answered in time. question_type forces QM or QU
+        questions instead of the automatic choice.
+        """
         info = ServiceInfo(type_, name)
         if info.request(self, timeout, question_type):
             return info
@@ -331,6 +339,12 @@ def register_service(
         cooperating_responders: bool = False,
         strict: bool = True,
     ) -> None:
+        """Announce a service on the network.
+
+        Probes for conflicts first; with allow_name_change the instance
+        name is renamed until it is unique. May raise EventLoopBlocked
+        when the event loop cannot complete the registration in time.
+        """
         assert self.loop is not None
         run_coro_with_timeout(
             await_awaitable(
@@ -348,6 +362,12 @@ async def async_register_service(
         cooperating_responders: bool = False,
         strict: bool = True,
     ) -> Awaitable:
+        """Announce a service on the network from the event loop.
+
+        Returns an awaitable that completes once the announcements have
+        been sent. Prefer setting TTLs on the ServiceInfo over the ttl
+        argument.
+        """
         if ttl is not None:
             # ttl argument is used to maintain backward compatibility
             # Setting TTLs via ServiceInfo is preferred
@@ -361,6 +381,11 @@ async def async_register_service(
         return asyncio.ensure_future(self._async_broadcast_service(info, _REGISTER_TIME, None))
 
     def update_service(self, info: ServiceInfo) -> None:
+        """Publish updated records for an already registered service.
+
+        May raise EventLoopBlocked when the event loop cannot complete
+        the update in time.
+        """
         assert self.loop is not None
         run_coro_with_timeout(
             await_awaitable(self.async_update_service(info)),
@@ -369,6 +394,10 @@ def update_service(self, info: ServiceInfo) -> None:
         )
 
     async def async_update_service(self, info: ServiceInfo) -> Awaitable:
+        """Publish updated records for an already registered service.
+
+        Returns an awaitable that completes once the rebroadcasts finish.
+        """
         self.registry.async_update(info)
         return asyncio.ensure_future(self._async_broadcast_service(info, _REGISTER_TIME, None))
 
@@ -464,6 +493,13 @@ async def async_get_service_info(
         timeout: int = 3000,
         question_type: DNSQuestionType | None = None,
     ) -> AsyncServiceInfo | None:
+        """Look up details for a service on the network from the event loop.
+
+        Queries for the given fully qualified type and name, waiting up to
+        timeout milliseconds, and returns a populated AsyncServiceInfo or
+        None when nothing answered in time. question_type forces QM or QU
+        questions instead of the automatic choice.
+        """
         info = AsyncServiceInfo(type_, name)
         if await info.async_request(self, timeout, question_type):
             return info

diff --git a/src/zeroconf/_dns.py b/src/zeroconf/_dns.py
@@ -158,6 +158,8 @@ def __repr__(self) -> str:
 
 
 class DNSRecord(DNSEntry):  # noqa: PLW1641
+    """A DNS entry that also carries a TTL and creation time."""
+
     __slots__ = ("created", "ttl")
 
     def __init__(
@@ -201,10 +203,12 @@ def get_expiration_time(self, percent: _int) -> float:
 
     # TODO: Switch to just int here
     def get_remaining_ttl(self, now: _float) -> int | float:
+        """Seconds of TTL left at the given time, never negative."""
         remain = (self.created + (_EXPIRE_FULL_TIME_MS * self.ttl) - now) / 1000.0
         return 0 if remain < 0 else remain
 
     def is_expired(self, now: _float) -> bool:
+        """True once the full TTL has elapsed."""
         return self.created + (_EXPIRE_FULL_TIME_MS * self.ttl) <= now
 
     def is_stale(self, now: _float) -> bool:
@@ -318,10 +322,12 @@ def _fast_init(
         self._hash = hash((self.key, type_, self.class_, cpu, os))
 
     def write(self, out: DNSOutgoing) -> None:
+        """Write the rdata to an outgoing packet."""
         out.write_character_string(self.cpu.encode("utf-8"))
         out.write_character_string(self.os.encode("utf-8"))
 
     def __eq__(self, other: Any) -> bool:
+        """Equal when cpu, os and the entry fields match."""
         return isinstance(other, DNSHinfo) and self._eq(other)
 
     def _eq(self, other: DNSHinfo) -> bool:
@@ -480,6 +486,7 @@ def write(self, out: DNSOutgoing) -> None:
         out.write_name(self.server)
 
     def __eq__(self, other: Any) -> bool:
+        """Equal when priority, weight, port, server and the entry fields match."""
         return isinstance(other, DNSService) and self._eq(other)
 
     def _eq(self, other: DNSService) -> bool:
@@ -533,6 +540,7 @@ def _fast_init(
         self._hash = hash((self.key, type_, self.class_, next_name, *self.rdtypes))
 
     def write(self, out: DNSOutgoing) -> None:
+        """Write the rdata to an outgoing packet."""
         bitmap = bytearray(b"\0" * 32)
         total_octets = 0
         for rdtype in self.rdtypes:
@@ -604,6 +612,7 @@ def _get_lookup(self) -> dict[DNSRecord, DNSRecord]:
         return self._lookup
 
     def suppresses(self, record: _DNSRecord) -> bool:
+        """True when the set holds a match with over half the record's TTL left."""
         lookup = self._get_lookup()
         other = lookup.get(record)
         if other is None:

diff --git a/src/zeroconf/_handlers/query_handler.py b/src/zeroconf/_handlers/query_handler.py
@@ -318,6 +318,7 @@ def _answer_question(
     def async_response(  # pylint: disable=unused-argument
         self, msgs: list[_DNSIncoming], ucast_source: bool
     ) -> QuestionAnswers | None:
+        """Build the answers for a batch of incoming queries, or None when there are none."""
         strategies: list[_AnswerStrategy] = []
         for msg in msgs:
             for question in msg._questions:

diff --git a/src/zeroconf/_protocol/incoming.py b/src/zeroconf/_protocol/incoming.py
@@ -140,9 +140,11 @@ def __reduce__(self) -> tuple:
         raise TypeError(f"cannot pickle {type(self).__name__!r} object")
 
     def is_query(self) -> bool:
+        """True when the QR flag marks the message as a query."""
         return (self.flags & _FLAGS_QR_MASK) == _FLAGS_QR_QUERY
 
     def is_response(self) -> bool:
+        """True when the QR flag marks the message as a response."""
         return (self.flags & _FLAGS_QR_MASK) == _FLAGS_QR_RESPONSE
 
     def has_qu_question(self) -> bool:
@@ -151,6 +153,7 @@ def has_qu_question(self) -> bool:
 
     @property
     def truncated(self) -> bool:
+        """True when the TC bit is set."""
         return (self.flags & _FLAGS_TC) == _FLAGS_TC
 
     @property
@@ -211,6 +214,7 @@ def answers(self) -> list[DNSRecord]:
         return self._answers
 
     def is_probe(self) -> bool:
+        """True when the message carries authority records, marking a probe."""
         return self._num_authorities > 0
 
     def __repr__(self) -> str:
@@ -270,6 +274,7 @@ def _read_questions(self) -> None:
             questions.append(question)
 
     def _read_character_string(self) -> str:
+        """Decode a length prefixed character string."""
         if self.offset >= self._data_len:
             raise IncomingDecodeError(
                 f"Character string at offset {self.offset} overruns packet of "
@@ -293,6 +298,7 @@ def _read_character_string(self) -> str:
         return info
 
     def _read_string(self, length: _int) -> bytes:
+        """Slice the next length bytes out of the buffer."""
         start = self.offset
         end = start + length
         if end > self._data_len:
@@ -481,6 +487,7 @@ def _read_bitmap(self, end: _int) -> list[int]:
         return rdtypes
 
     def _read_name(self) -> str:
+        """Decode a possibly compressed domain name at the current offset."""
         original_offset = self.offset
         name_str_cache = self._name_str_cache
         is_pure_pointer = False