Skip to content

Cleanup typing in zeroconf._services #711

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bdraco merged 1 commit into from
Jun 17, 2021
+31 −32
Merged

Cleanup typing in zeroconf._services #711

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 31 additions & 32 deletions zeroconf/_services/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ def update_service(self, zc: 'Zeroconf', type_: str, name: str) -> None:

class Signal:
def __init__(self) -> None:
self._handlers = [] # type: List[Callable[..., None]]
self._handlers: List[Callable[..., None]] = []

def fire(self, **kwargs: Any) -> None:
for h in list(self._handlers):
Expand Down Expand Up @@ -233,7 +233,7 @@ def __init__(
) -> None:
"""Creates a browser for a specific type"""
assert handlers or listener, 'You need to specify at least one handler'
self.types = set(type_ if isinstance(type_, list) else [type_]) # type: Set[str]
self.types: Set[str] = set(type_ if isinstance(type_, list) else [type_])
for check_type_ in self.types:
# Will generate BadTypeInNameException on a bad name
service_type_name(check_type_, strict=False)
Expand All @@ -245,9 +245,8 @@ def __init__(
current_time = current_time_millis()
self._next_time = {check_type_: current_time for check_type_ in self.types}
self._delay = {check_type_: delay for check_type_ in self.types}
self._pending_handlers = OrderedDict() # type: OrderedDict[Tuple[str, str], ServiceStateChange]
self._handlers_to_call = OrderedDict() # type: OrderedDict[Tuple[str, str], ServiceStateChange]

self._pending_handlers: OrderedDict[Tuple[str, str], ServiceStateChange] = OrderedDict()
self._handlers_to_call: OrderedDict[Tuple[str, str], ServiceStateChange] = OrderedDict()
self._service_state_changed = Signal()

self.done = False
Expand Down Expand Up @@ -552,13 +551,13 @@ def __init__(
self.port = port
self.weight = weight
self.priority = priority
if server:
self.server = server
else:
self.server = name
self.server = server if server else name
self.server_key = self.server.lower()
self._properties = {} # type: Dict
self._set_properties(properties)
self._properties: Dict[Union[str, bytes], Optional[Union[str, bytes]]] = {}
if isinstance(properties, bytes):
self._set_text(properties)
else:
self._set_properties(properties)
self.host_ttl = host_ttl
self.other_ttl = other_ttl

Expand Down Expand Up @@ -618,33 +617,33 @@ def parsed_addresses(self, version: IPVersion = IPVersion.All) -> List[str]:
for addr in result
]

def _set_properties(self, properties: Union[bytes, Dict]) -> None:
def _set_properties(self, properties: Dict) -> None:
"""Sets properties and text of this info from a dictionary"""
if isinstance(properties, dict):
self._properties = properties
list_ = []
result = b''
for key, value in properties.items():
if isinstance(key, str):
key = key.encode('utf-8')

record = key
if value is not None:
if not isinstance(value, bytes):
value = str(value).encode('utf-8')
record += b'=' + value
list_.append(record)
for item in list_:
result = b''.join((result, int2byte(len(item)), item))
self.text = result
else:
self.text = properties
self._properties = properties
list_ = []
result = b''
for key, value in properties.items():
if isinstance(key, str):
key = key.encode('utf-8')

record = key
if value is not None:
if not isinstance(value, bytes):
value = str(value).encode('utf-8')
record += b'=' + value
list_.append(record)
for item in list_:
result = b''.join((result, int2byte(len(item)), item))
self.text = result

def _set_text(self, text: bytes) -> None:
"""Sets properties and text given a text field"""
self.text = text
result = {} # type: Dict
end = len(text)
if end == 0:
self._properties = {}
return
result: Dict[Union[str, bytes], Optional[Union[str, bytes]]] = {}
index = 0
strs = []
while index < end:
Expand Down