python-zeroconf · bdraco · Aug 28, 2026 · Aug 26, 2026 · Aug 26, 2026
diff --git a/src/zeroconf/_cache.py b/src/zeroconf/_cache.py
@@ -42,6 +42,7 @@
 _UNIQUE_RECORD_TYPES = (DNSAddress, DNSHinfo, DNSPointer, DNSText, DNSService)
 _UniqueRecordsType = DNSAddress | DNSHinfo | DNSPointer | DNSText | DNSService
 _DNSRecordCacheType = dict[str, dict[DNSRecord, DNSRecord]]
+_UniqueType = tuple[str, int, int]
 _DNSRecord = DNSRecord
 _str = str
 _float = float
@@ -324,7 +325,7 @@ def names(self) -> list[str]:
 
     def async_mark_unique_records_older_than_1s_to_expire(
         self,
-        unique_types: set[tuple[_str, _int, _int]],
+        unique_types: set[_UniqueType],
         answers: Iterable[DNSRecord],
         now: _float,
     ) -> None:

diff --git a/src/zeroconf/_handlers/query_handler.pxd b/src/zeroconf/_handlers/query_handler.pxd
@@ -96,7 +96,6 @@ cdef class QueryHandler:
         question=DNSQuestion,
         answer_set=cython.dict,
         known_answers=DNSRRSet,
-        known_answers_set=cython.set,
         is_unicast=bint,
         is_probe=object,
         now=double

diff --git a/src/zeroconf/_handlers/query_handler.py b/src/zeroconf/_handlers/query_handler.py
@@ -61,6 +61,8 @@
 
 _int = int
 _str = str
+_ServiceInfo = ServiceInfo
+_DNSIncoming = DNSIncoming
 
 _ANSWER_STRATEGY_SERVICE_TYPE_ENUMERATION = 0
 _ANSWER_STRATEGY_POINTER = 1
@@ -212,7 +214,7 @@ def __init__(self, zc: Zeroconf) -> None:
 
     def _add_service_type_enumeration_query_answers(
         self,
-        types: list[str],
+        types: list[_str],
         answer_set: _AnswerWithAdditionalsType,
         known_answers: DNSRRSet,
     ) -> None:
@@ -234,7 +236,7 @@ def _add_service_type_enumeration_query_answers(
 
     def _add_pointer_answers(
         self,
-        services: list[ServiceInfo],
+        services: list[_ServiceInfo],
         answer_set: _AnswerWithAdditionalsType,
         known_answers: DNSRRSet,
     ) -> None:
@@ -253,7 +255,7 @@ def _add_pointer_answers(
 
     def _add_address_answers(
         self,
-        services: list[ServiceInfo],
+        services: list[_ServiceInfo],
         answer_set: _AnswerWithAdditionalsType,
         known_answers: DNSRRSet,
         type_: _int,
@@ -284,8 +286,8 @@ def _answer_question(
         self,
         question: DNSQuestion,
         strategy_type: _int,
-        types: list[str],
-        services: list[ServiceInfo],
+        types: list[_str],
+        services: list[_ServiceInfo],
         known_answers: DNSRRSet,
     ) -> _AnswerWithAdditionalsType:
         """Answer a question."""
@@ -313,7 +315,7 @@ def _answer_question(
         return answer_set
 
     def async_response(  # pylint: disable=unused-argument
-        self, msgs: list[DNSIncoming], ucast_source: bool
+        self, msgs: list[_DNSIncoming], ucast_source: bool
     ) -> QuestionAnswers | None:
         """Deal with incoming query packets. Provides a response if possible.
 
@@ -435,7 +437,7 @@ def _get_answer_strategies(
 
     def handle_assembled_query(
         self,
-        packets: list[DNSIncoming],
+        packets: list[_DNSIncoming],
         addr: _str,
         port: _int,
         transport: _WrappedTransport,

diff --git a/src/zeroconf/_handlers/record_manager.py b/src/zeroconf/_handlers/record_manager.py
@@ -37,6 +37,8 @@
     from .._core import Zeroconf
 
 _float = float
+_RecordUpdate = RecordUpdate
+_DNSQuestion = DNSQuestion
 
 
 class RecordManager:
@@ -50,7 +52,7 @@ def __init__(self, zeroconf: Zeroconf) -> None:
         self.cache = zeroconf.cache
         self.listeners: set[RecordUpdateListener] = set()
 
-    def async_updates(self, now: _float, records: list[RecordUpdate]) -> None:
+    def async_updates(self, now: _float, records: list[_RecordUpdate]) -> None:
         """Used to notify listeners of new information that has updated
         a record.
 
@@ -190,7 +192,7 @@ def async_add_listener(
         self._async_update_matching_records(listener, questions)
 
     def _async_update_matching_records(
-        self, listener: RecordUpdateListener, questions: list[DNSQuestion]
+        self, listener: RecordUpdateListener, questions: list[_DNSQuestion]
     ) -> None:
         """Calls back any existing entries in the cache that answer the question.
 

diff --git a/src/zeroconf/_protocol/incoming.py b/src/zeroconf/_protocol/incoming.py
@@ -455,7 +455,7 @@ def _read_name(self) -> str:
         return name
 
     def _decode_labels_at_offset(
-        self, off: _int, labels: list[str], seen_pointers: set[int], depth: _int
+        self, off: _int, labels: list[_str], seen_pointers: set[_int], depth: _int
     ) -> int:
         # This is a tight loop that is called frequently, small optimizations can make a difference.
         if depth > MAX_DNS_LABELS:

diff --git a/src/zeroconf/_services/browser.pxd b/src/zeroconf/_services/browser.pxd
@@ -56,7 +56,7 @@ cpdef list generate_service_query(
 )
 
 
-@cython.locals(answer=DNSPointer, query_buckets=list, question=DNSQuestion, max_compressed_size=cython.uint, max_bucket_size=cython.uint, query_bucket=_DNSPointerOutgoingBucket)
+@cython.locals(answer=DNSPointer, question=DNSQuestion, max_compressed_size=cython.uint, max_bucket_size=cython.uint, query_bucket=_DNSPointerOutgoingBucket)
 cdef list _group_ptr_queries_with_known_answers(double now_millis, bint multicast, cython.dict question_with_known_answers)
 
 
@@ -92,7 +92,7 @@ cdef class QueryScheduler:
 
     cpdef void _process_startup_queries(self)
 
-    @cython.locals(query=_ScheduledPTRQuery, next_scheduled=_ScheduledPTRQuery, next_when=double)
+    @cython.locals(query=_ScheduledPTRQuery, next_when=double)
     cpdef void _process_ready_types(self)
 
     cpdef void async_send_ready_queries(self, bint first_request, double now_millis, set ready_types)

diff --git a/src/zeroconf/_services/browser.py b/src/zeroconf/_services/browser.py
@@ -95,6 +95,9 @@
 str_ = str
 
 _QuestionWithKnownAnswers = dict[DNSQuestion, set[DNSPointer]]
+DNSPointer_ = DNSPointer
+DNSOutgoing_ = DNSOutgoing
+RecordUpdate_ = RecordUpdate
 
 heappop = heapq.heappop
 heappush = heapq.heappush
@@ -193,7 +196,7 @@ def __init__(self, now_millis: float, multicast: bool) -> None:
         self.out = DNSOutgoing(_FLAGS_QR_QUERY, multicast)
         self.bytes = 0
 
-    def add(self, max_compressed_size: int_, question: DNSQuestion, answers: set[DNSPointer]) -> None:
+    def add(self, max_compressed_size: int_, question: DNSQuestion, answers: set[DNSPointer_]) -> None:
         """Add a new set of questions and known answers to the outgoing."""
         self.out.add_question(question)
         for answer in answers:
@@ -259,10 +262,10 @@ def _group_ptr_queries_with_known_answers(
 def generate_service_query(
     zc: Zeroconf,
     now_millis: float_,
-    types_: set[str],
+    types_: set[str_],
     multicast: bool,
     question_type: DNSQuestionType | None,
-) -> list[DNSOutgoing]:
+) -> list[DNSOutgoing_]:
     """Generate a service query for sending with zeroconf.send."""
     questions_with_known_answers: _QuestionWithKnownAnswers = {}
     qu_question = not multicast if question_type is None else question_type is QU_QUESTION
@@ -533,7 +536,7 @@ def _process_ready_types(self) -> None:
         self._next_run = self._loop.call_at(millis_to_seconds(next_when_millis), self._process_ready_types)
 
     def async_send_ready_queries(
-        self, first_request: bool, now_millis: float_, ready_types: set[str]
+        self, first_request: bool, now_millis: float_, ready_types: set[str_]
     ) -> None:
         """Send any ready queries."""
         # If they did not specify and this is the first request, ask QU questions
@@ -665,7 +668,7 @@ def _enqueue_callback(
         ):
             self._pending_handlers[key] = state_change
 
-    def async_update_records(self, zc: Zeroconf, now: float_, records: list[RecordUpdate]) -> None:
+    def async_update_records(self, zc: Zeroconf, now: float_, records: list[RecordUpdate_]) -> None:
         """Callback invoked by Zeroconf when new information arrives.
 
         Updates information required by browser in the Zeroconf cache.

diff --git a/src/zeroconf/_services/info.py b/src/zeroconf/_services/info.py
@@ -99,6 +99,7 @@
 float_ = float
 int_ = int
 str_ = str
+RecordUpdate_ = RecordUpdate
 
 QU_QUESTION = DNSQuestionType.QU
 QM_QUESTION = DNSQuestionType.QM
@@ -547,7 +548,7 @@ def _set_ipv4_addresses_from_cache(self, zc: Zeroconf, now: float_) -> None:
         else:
             self._ipv4_addresses = self._get_ip_addresses_from_cache_lifo(zc, now, _TYPE_A)
 
-    def async_update_records(self, zc: Zeroconf, now: float_, records: list[RecordUpdate]) -> None:
+    def async_update_records(self, zc: Zeroconf, now: float_, records: list[RecordUpdate_]) -> None:
         """Updates service information from a DNS record.
 
         This method will be run in the event loop.
@@ -741,7 +742,7 @@ def dns_nsec(self, missing_types: list[int], override_ttl: int_ | None = None) -
         """Return DNSNsec from ServiceInfo."""
         return self._dns_nsec(missing_types, override_ttl)
 
-    def _dns_nsec(self, missing_types: list[int], override_ttl: int_ | None) -> DNSNsec:
+    def _dns_nsec(self, missing_types: list[int_], override_ttl: int_ | None) -> DNSNsec:
         """Return DNSNsec from ServiceInfo."""
         return DNSNsec(
             self._name,

diff --git a/src/zeroconf/_services/registry.py b/src/zeroconf/_services/registry.py
@@ -26,6 +26,7 @@
 from .info import ServiceInfo
 
 _str = str
+_ServiceInfo = ServiceInfo
 
 
 class ServiceRegistry:
@@ -79,7 +80,7 @@ def async_get_infos_server(self, server: str) -> list[ServiceInfo]:
         """Return all ServiceInfo matching server."""
         return self._async_get_by_index(self.servers, server)
 
-    def _async_get_by_index(self, records: dict[str, list], key: _str) -> list[ServiceInfo]:
+    def _async_get_by_index(self, records: dict[_str, list], key: _str) -> list[_ServiceInfo]:
         """Return all ServiceInfo matching the index."""
         record_list = records.get(key)
         if record_list is None:
@@ -98,7 +99,7 @@ def _add(self, info: ServiceInfo) -> None:
         self.servers.setdefault(info.server_key, []).append(info.key)
         self.has_entries = True
 
-    def _remove(self, infos: list[ServiceInfo]) -> None:
+    def _remove(self, infos: list[_ServiceInfo]) -> None:
         """Remove a services under the lock."""
         for info in infos:
             old_service_info = self._services.get(info.key)