python-zeroconf · bdraco · Jun 18, 2023 · Jun 18, 2023 · Jun 18, 2023
diff --git a/src/zeroconf/_cache.py b/src/zeroconf/_cache.py
@@ -144,7 +144,7 @@ def async_all_by_details(self, name: _str, type_: int, class_: int) -> Iterable[
         """
         return self._async_all_by_details(name, type_, class_)
 
-    def _async_all_by_details(self, name: _str, type_: int, class_: int) -> List[DNSRecord]:
+    def _async_all_by_details(self, name: _str, type_: _int, class_: _int) -> List[DNSRecord]:
         """Gets all matching entries by details.
 
         This function is not thread-safe and must be called from
@@ -258,5 +258,5 @@ def _async_mark_unique_records_older_than_1s_to_expire(
                     record.set_created_ttl(now, 1)
 
 
-def _dns_record_matches(record: _DNSRecord, key: _str, type_: int, class_: int) -> bool:
+def _dns_record_matches(record: _DNSRecord, key: _str, type_: _int, class_: _int) -> bool:
     return key == record.key and type_ == record.type and class_ == record.class_
diff --git a/src/zeroconf/_protocol/incoming.py b/src/zeroconf/_protocol/incoming.py
@@ -67,6 +67,7 @@
 
 _seen_logs: Dict[str, Union[int, tuple]] = {}
 _str = str
+_int = int
 
 
 class DNSIncoming:
@@ -231,7 +232,7 @@ def _read_character_string(self) -> bytes:
         self.offset += 1
         return self._read_string(length)
 
-    def _read_string(self, length: int) -> bytes:
+    def _read_string(self, length: _int) -> bytes:
         """Reads a string of a given length from the packet"""
         info = self.data[self.offset : self.offset + length]
         self.offset += length
@@ -267,7 +268,7 @@ def _read_others(self) -> None:
                 self._answers.append(rec)
 
     def _read_record(
-        self, domain: _str, type_: int, class_: int, ttl: int, length: int
+        self, domain: _str, type_: _int, class_: _int, ttl: _int, length: _int
     ) -> Optional[DNSRecord]:
         """Read known records types and skip unknown ones."""
         if type_ == _TYPE_A:
@@ -324,7 +325,7 @@ def _read_record(
         self.offset += length
         return None
 
-    def _read_bitmap(self, end: int) -> List[int]:
+    def _read_bitmap(self, end: _int) -> List[int]:
         """Reads an NSEC bitmap from the packet."""
         rdtypes = []
         while self.offset < end:
@@ -355,7 +356,7 @@ def _read_name(self) -> str:
             )
         return name
 
-    def _decode_labels_at_offset(self, off: int, labels: List[str], seen_pointers: Set[int]) -> int:
+    def _decode_labels_at_offset(self, off: _int, labels: List[str], seen_pointers: Set[int]) -> int:
         # This is a tight loop that is called frequently, small optimizations can make a difference.
         while off < self._data_len:
             length = self.data[off]

diff --git a/src/zeroconf/_protocol/outgoing.py b/src/zeroconf/_protocol/outgoing.py
@@ -42,6 +42,7 @@
 
 str_ = str
 float_ = float
+int_ = int
 DNSQuestion_ = DNSQuestion
 DNSRecord_ = DNSRecord
 
@@ -197,20 +198,20 @@ def add_question_or_all_cache(
         for cached_entry in cached_entries:
             self.add_answer_at_time(cached_entry, now)
 
-    def _write_byte(self, value: int) -> None:
+    def _write_byte(self, value: int_) -> None:
         """Writes a single byte to the packet"""
         self.data.append(value.to_bytes(1, 'big'))
         self.size += 1
 
-    def _insert_short_at_start(self, value: int) -> None:
+    def _insert_short_at_start(self, value: int_) -> None:
         """Inserts an unsigned short at the start of the packet"""
         self.data.insert(0, value.to_bytes(2, 'big'))
 
-    def _replace_short(self, index: int, value: int) -> None:
+    def _replace_short(self, index: int_, value: int_) -> None:
         """Replaces an unsigned short in a certain position in the packet"""
         self.data[index] = value.to_bytes(2, 'big')
 
-    def write_short(self, value: int) -> None:
+    def write_short(self, value: int_) -> None:
         """Writes an unsigned short to the packet"""
         self.data.append(value.to_bytes(2, 'big'))
         self.size += 2
@@ -321,7 +322,7 @@ def _write_record(self, record: DNSRecord_, now: float_) -> bool:
         self._replace_short(index, length)
         return self._check_data_limit_or_rollback(start_data_length, start_size)
 
-    def _check_data_limit_or_rollback(self, start_data_length: int, start_size: int) -> bool:
+    def _check_data_limit_or_rollback(self, start_data_length: int_, start_size: int_) -> bool:
         """Check data limit, if we go over, then rollback and return False."""
         len_limit = _MAX_MSG_ABSOLUTE if self.allow_long else _MAX_MSG_TYPICAL
         self.allow_long = False
@@ -338,23 +339,23 @@ def _check_data_limit_or_rollback(self, start_data_length: int, start_size: int)
             del self.names[name]
         return False
 
-    def _write_questions_from_offset(self, questions_offset: int) -> int:
+    def _write_questions_from_offset(self, questions_offset: int_) -> int:
         questions_written = 0
         for question in self.questions[questions_offset:]:
             if not self._write_question(question):
                 break
             questions_written += 1
         return questions_written
 
-    def _write_answers_from_offset(self, answer_offset: int) -> int:
+    def _write_answers_from_offset(self, answer_offset: int_) -> int:
         answers_written = 0
         for answer, time_ in self.answers[answer_offset:]:
             if not self._write_record(answer, time_):
                 break
             answers_written += 1
         return answers_written
 
-    def _write_records_from_offset(self, records: Sequence[DNSRecord], offset: int) -> int:
+    def _write_records_from_offset(self, records: Sequence[DNSRecord], offset: int_) -> int:
         records_written = 0
         for record in records[offset:]:
             if not self._write_record(record, 0):
@@ -363,7 +364,7 @@ def _write_records_from_offset(self, records: Sequence[DNSRecord], offset: int)
         return records_written
 
     def _has_more_to_add(
-        self, questions_offset: int, answer_offset: int, authority_offset: int, additional_offset: int
+        self, questions_offset: int_, answer_offset: int_, authority_offset: int_, additional_offset: int_
     ) -> bool:
         """Check if all questions, answers, authority, and additionals have been written to the packet."""
         return (