Select all rules for Ruff (#1106)

Author: PerchunPak

docs/conf.py

@@ -15,12 +15,11 @@
 
 import os
 import sys
-
-from packaging.version import Version, parse as parse_version
 from importlib.metadata import version as importlib_version
 
+from packaging.version import Version, parse as parse_version
 
-sys.path.insert(0, os.path.abspath(".."))
+sys.path.insert(0, os.path.abspath(".."))  # noqa: PTH100
 
 
 # -- Project information -----------------------------------------------------
@@ -132,7 +131,7 @@ def mock_autodoc() -> None:
 
     See also https://stackoverflow.com/a/75041544/20952782.
     """
-    from sphinx.ext import autodoc
+    from sphinx.ext import autodoc  # noqa: PLC0415
 
     class MockedClassDocumenter(autodoc.ClassDocumenter):
         def add_line(self, line: str, source: str, *lineno: int) -> None:
@@ -147,8 +146,8 @@ def delete_doc_for_address_base() -> None:
     """``__new__`` is appended to the docstring of ``AddressBase``.
 
     And we do not want autogenerated nonsense docstring there.
-    """
-    from mcstatus.address import _AddressBase
+    """  # noqa: D401 # imperative mood
+    from mcstatus.address import _AddressBase  # noqa: PLC0415
 
     del _AddressBase.__new__.__doc__
 

docs/examples/code/ping_as_java_and_bedrock_in_one_time.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+
 import asyncio
 
 from mcstatus import BedrockServer, JavaServer
@@ -31,7 +32,7 @@ async def status(host: str) -> JavaStatusResponse | BedrockStatusResponse:
 async def handle_exceptions(done: set[asyncio.Task], pending: set[asyncio.Task]) -> asyncio.Task | None:
     """Handle exceptions from tasks.
 
-    Also, cancel all pending tasks, if found correct one.
+    Also, cancel all pending tasks, if found the correct one.
     """
     if len(done) == 0:
         raise ValueError("No tasks was given to `done` set.")
@@ -47,14 +48,15 @@ async def handle_exceptions(done: set[asyncio.Task], pending: set[asyncio.Task])
             for pending_task in pending:
                 pending_task.cancel()
             return task
+    return None
 
 
 async def handle_java(host: str) -> JavaStatusResponse:
-    """A wrapper around mcstatus, to compress it in one function."""
+    """Wrap mcstatus, to compress lookup and status into one function."""
     return await (await JavaServer.async_lookup(host)).async_status()
 
 
 async def handle_bedrock(host: str) -> BedrockStatusResponse:
-    """A wrapper around mcstatus, to compress it in one function."""
+    """Wrap mcstatus, to compress lookup and status into one function."""
     # note: `BedrockServer` doesn't have `async_lookup` method, see it's docstring
     return await BedrockServer.lookup(host).async_status()

mcstatus/__main__.py

@@ -1,16 +1,20 @@
+# ruff: noqa: T201 # usage of `print`
 from __future__ import annotations
 
-import dns.resolver
-import sys
-import json
 import argparse
-import socket
 import dataclasses
-from typing import TypeAlias
+import json
+import socket
+import sys
+from typing import TYPE_CHECKING, TypeAlias
+
+import dns.resolver
 
-from mcstatus import JavaServer, LegacyServer, BedrockServer
+from mcstatus import BedrockServer, JavaServer, LegacyServer
 from mcstatus.responses import JavaStatusResponse
-from mcstatus.motd import Motd
+
+if TYPE_CHECKING:
+    from mcstatus.motd import Motd
 
 SupportedServers: TypeAlias = "JavaServer | LegacyServer | BedrockServer"
 
@@ -30,21 +34,19 @@
 
 
 def _motd(motd: Motd) -> str:
-    """Formats MOTD for human-readable output, with leading line break
-    if multiline."""
+    """Format MOTD for human-readable output, with leading line break if multiline."""
     s = motd.to_ansi()
     return f"\n{s}" if "\n" in s else f" {s}"
 
 
 def _kind(serv: SupportedServers) -> str:
     if isinstance(serv, JavaServer):
         return "Java"
-    elif isinstance(serv, LegacyServer):
+    if isinstance(serv, LegacyServer):
         return "Java (pre-1.7)"
-    elif isinstance(serv, BedrockServer):
+    if isinstance(serv, BedrockServer):
         return "Bedrock"
-    else:
-        raise ValueError(f"unsupported server for kind: {serv}")
+    raise ValueError(f"unsupported server for kind: {serv}")
 
 
 def _ping_with_fallback(server: SupportedServers) -> float:
@@ -56,7 +58,7 @@ def _ping_with_fallback(server: SupportedServers) -> float:
     ping_exc = None
     try:
         return server.ping(tries=1)
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 # blindly catching Exception
         ping_exc = e
 
     latency = server.status().latency
@@ -103,13 +105,13 @@ def json_cmd(server: SupportedServers) -> int:
     status_res = query_res = exn = None
     try:
         status_res = server.status(tries=1)
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 # blindly catching Exception
         exn = exn or e
 
     try:
         if isinstance(server, JavaServer):
             query_res = server.query(tries=1)
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 # blindly catching Exception
         exn = exn or e
 
     # construct 'data' dict outside try/except to ensure data processing errors
@@ -147,7 +149,7 @@ def query_cmd(server: SupportedServers) -> int:
 
     try:
         response = server.query()
-    except socket.timeout:
+    except TimeoutError:
         print(QUERY_FAIL_WARNING, file=sys.stderr)
         return 1
 
@@ -199,7 +201,7 @@ def main(argv: list[str] = sys.argv[1:]) -> int:
     try:
         server = lookup(args.address)
         return args.func(server)
-    except (socket.timeout, socket.gaierror, dns.resolver.NoNameservers, ConnectionError, TimeoutError) as e:
+    except (socket.gaierror, dns.resolver.NoNameservers, ConnectionError, TimeoutError) as e:
         # catch and hide traceback for expected user-facing errors
         print(f"Error: {e!r}", file=sys.stderr)
         return 1

mcstatus/address.py

@@ -3,7 +3,6 @@
 import ipaddress
 import sys
 import warnings
-from pathlib import Path
 from typing import NamedTuple, TYPE_CHECKING
 from urllib.parse import urlparse
 
@@ -12,14 +11,16 @@
 import mcstatus.dns
 
 if TYPE_CHECKING:
+    from pathlib import Path
+
     from typing_extensions import Self
 
 
 __all__ = ("Address", "async_minecraft_srv_address_lookup", "minecraft_srv_address_lookup")
 
 
 def _valid_urlparse(address: str) -> tuple[str, int | None]:
-    """Parses a string address like 127.0.0.1:25565 into host and port parts
+    """Parse a string address like 127.0.0.1:25565 into host and port parts.
 
     If the address doesn't have a specified port, None will be returned instead.
 
@@ -57,7 +58,7 @@ class Address(_AddressBase):
         The class is not a part of a Public API, but attributes :attr:`host` and :attr:`port` are a part of Public API.
     """
 
-    def __init__(self, host: str, port: int):
+    def __init__(self, host: str, port: int):  # noqa: ARG002 # unused arguments
         # We don't pass the host & port args to super's __init__, because NamedTuples handle
         # everything from __new__ and the passed self already has all of the parameters set.
         super().__init__()
@@ -93,7 +94,7 @@ def from_path(cls, path: Path, *, default_port: int | None = None) -> Self:
 
     @classmethod
     def parse_address(cls, address: str, *, default_port: int | None = None) -> Self:
-        """Parses a string address like ``127.0.0.1:25565`` into :attr:`.host` and :attr:`.port` parts.
+        """Parse a string address like ``127.0.0.1:25565`` into :attr:`.host` and :attr:`.port` parts.
 
         If the address has a port specified, use it, if not, fall back to ``default_port`` kwarg.
 
@@ -112,7 +113,7 @@ def parse_address(cls, address: str, *, default_port: int | None = None) -> Self
         return cls(host=hostname, port=port)
 
     def resolve_ip(self, lifetime: float | None = None) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
-        """Resolves a hostname's A record into an IP address.
+        """Resolve a hostname's A record into an IP address.
 
         If the host is already an IP, this resolving is skipped
         and host is returned directly.
@@ -150,7 +151,7 @@ def resolve_ip(self, lifetime: float | None = None) -> ipaddress.IPv4Address | i
         return self._cached_ip
 
     async def async_resolve_ip(self, lifetime: float | None = None) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
-        """Resolves a hostname's A record into an IP address.
+        """Resolve a hostname's A record into an IP address.
 
         See the docstring for :meth:`.resolve_ip` for further info. This function is purely
         an async alternative to it.

mcstatus/bedrock_status.py

@@ -4,12 +4,15 @@
 import socket
 import struct
 from time import perf_counter
+from typing import TYPE_CHECKING
 
 import asyncio_dgram
 
-from mcstatus.address import Address
 from mcstatus.responses import BedrockStatusResponse
 
+if TYPE_CHECKING:
+    from mcstatus.address import Address
+
 
 class BedrockServerStatus:
     request_status_data = bytes.fromhex(

mcstatus/dns.py

@@ -1,16 +1,18 @@
 from __future__ import annotations
 
-from typing import cast
+from typing import TYPE_CHECKING, cast
 
 import dns.asyncresolver
 import dns.resolver
 from dns.rdatatype import RdataType
-from dns.rdtypes.IN.A import A as ARecordAnswer
-from dns.rdtypes.IN.SRV import SRV as SRVRecordAnswer  # noqa: N811 # constant imported as non constant (it's class)
+
+if TYPE_CHECKING:
+    from dns.rdtypes.IN.A import A as ARecordAnswer
+    from dns.rdtypes.IN.SRV import SRV as SRVRecordAnswer  # noqa: N811 # constant imported as non constant (it's class)
 
 
 def resolve_a_record(hostname: str, lifetime: float | None = None) -> str:
-    """Perform a DNS resolution for an A record to given hostname
+    """Perform a DNS resolution for an A record to given hostname.
 
     :param hostname: The address to resolve for.
     :return: The resolved IP address from the A record
@@ -22,7 +24,7 @@ def resolve_a_record(hostname: str, lifetime: float | None = None) -> str:
     answers = dns.resolver.resolve(hostname, RdataType.A, lifetime=lifetime, search=True)
     # There should only be one answer here, though in case the server
     # does actually point to multiple IPs, we just pick the first one
-    answer = cast(ARecordAnswer, answers[0])
+    answer = cast("ARecordAnswer", answers[0])
     ip = str(answer).rstrip(".")
     return ip
 
@@ -35,7 +37,7 @@ async def async_resolve_a_record(hostname: str, lifetime: float | None = None) -
     answers = await dns.asyncresolver.resolve(hostname, RdataType.A, lifetime=lifetime, search=True)
     # There should only be one answer here, though in case the server
     # does actually point to multiple IPs, we just pick the first one
-    answer = cast(ARecordAnswer, answers[0])
+    answer = cast("ARecordAnswer", answers[0])
     ip = str(answer).rstrip(".")
     return ip
 
@@ -53,7 +55,7 @@ def resolve_srv_record(query_name: str, lifetime: float | None = None) -> tuple[
     answers = dns.resolver.resolve(query_name, RdataType.SRV, lifetime=lifetime, search=True)
     # There should only be one answer here, though in case the server
     # does actually point to multiple IPs, we just pick the first one
-    answer = cast(SRVRecordAnswer, answers[0])
+    answer = cast("SRVRecordAnswer", answers[0])
     host = str(answer.target).rstrip(".")
     port = int(answer.port)
     return host, port
@@ -67,7 +69,7 @@ async def async_resolve_srv_record(query_name: str, lifetime: float | None = Non
     answers = await dns.asyncresolver.resolve(query_name, RdataType.SRV, lifetime=lifetime, search=True)
     # There should only be one answer here, though in case the server
     # does actually point to multiple IPs, we just pick the first one
-    answer = cast(SRVRecordAnswer, answers[0])
+    answer = cast("SRVRecordAnswer", answers[0])
     host = str(answer.target).rstrip(".")
     port = int(answer.port)
     return host, port

mcstatus/forge_data.py

@@ -137,9 +137,7 @@ def decode(cls, buffer: Connection) -> tuple[Self, list[ForgeDataChannel]]:
         if not is_server:
             mod_version = buffer.read_utf()
 
-        channels: list[ForgeDataChannel] = []
-        for _ in range(channel_count):
-            channels.append(ForgeDataChannel.decode(buffer, mod_id))
+        channels = [ForgeDataChannel.decode(buffer, mod_id) for _ in range(channel_count)]
 
         return cls(name=mod_id, marker=mod_version), channels
 
@@ -161,7 +159,7 @@ def read(self, length: int) -> bytearray:
         while len(data) < length:
             result = self.stringio.read(1)
             if not result:
-                raise IOError(f"Not enough data to read! {len(data)} < {length}")
+                raise OSError(f"Not enough data to read! {len(data)} < {length}")
             data.extend(result.encode("utf-16be"))
         while len(data) > length:
             self.received.append(data.pop())
@@ -212,7 +210,7 @@ def _decode_optimized(string: str) -> Connection:
             return str_buffer.read_optimized_buffer()
 
     @classmethod
-    def build(cls, raw: RawForgeData) -> Self | None:
+    def build(cls, raw: RawForgeData) -> Self:
         """Build an object about Forge mods from raw response.
 
         :param raw: ``forgeData`` attribute in raw response :class:`dict`.
@@ -248,9 +246,8 @@ def build(cls, raw: RawForgeData) -> Self | None:
                 mods.append(mod)
 
             non_mod_channel_count = buffer.read_varint()
-            for _ in range(non_mod_channel_count):
-                channels.append(ForgeDataChannel.decode(buffer))
-        except IOError:
+            channels.extend(ForgeDataChannel.decode(buffer) for _ in range(non_mod_channel_count))
+        except OSError:
             if not truncated:
                 raise  # If answer wasn't truncated, we lost some data on the way
 

mcstatus/legacy_status.py

@@ -14,7 +14,7 @@ class _BaseLegacyServerStatus:
     def parse_response(data: bytes, latency: float) -> LegacyStatusResponse:
         decoded_data = data.decode("UTF-16BE").split("\0")
         if decoded_data[0] != "§1":
-            raise IOError("Received invalid kick packet reason")
+            raise OSError("Received invalid kick packet reason")
 
         return LegacyStatusResponse.build(decoded_data[1:], latency)
 
@@ -29,7 +29,7 @@ def read_status(self) -> LegacyStatusResponse:
         self.connection.write(self.request_status_data)
         id = self.connection.read(1)
         if id != b"\xff":
-            raise IOError("Received invalid packet ID")
+            raise OSError("Received invalid packet ID")
         length = self.connection.read_ushort()
         data = self.connection.read(length * 2)
         end = perf_counter()
@@ -46,7 +46,7 @@ async def read_status(self) -> LegacyStatusResponse:
         self.connection.write(self.request_status_data)
         id = await self.connection.read(1)
         if id != b"\xff":
-            raise IOError("Received invalid packet ID")
+            raise OSError("Received invalid packet ID")
         length = await self.connection.read_ushort()
         data = await self.connection.read(length * 2)
         end = perf_counter()