Modernize type hints and other ruff fixes

Author: cclauss

docs/source/conf.py

@@ -11,8 +11,8 @@
 # documentation root, use os.path.abspath to make it absolute, like shown here.
 #
 import os
-import sys
 import re
+import sys
 
 sys.path.insert(0, os.path.abspath("../.."))
 

setup.py

@@ -3,7 +3,7 @@
 import os
 import re
 
-from setuptools import setup, find_packages
+from setuptools import find_packages, setup
 
 PROJECT_ROOT = os.path.dirname(__file__)
 

src/priority/__init__.py

@@ -1,8 +1,8 @@
-# -*- coding: utf-8 -*-
 """
 priority: HTTP/2 priority implementation for Python
 """
-from .priority import (  # noqa
+
+from .priority import (  # noqa: F401
     Stream,
     PriorityTree,
     DeadlockError,

src/priority/priority.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """
 priority/tree
 ~~~~~~~~~~~~~
@@ -8,8 +7,6 @@
 
 import heapq
 
-from typing import List, Tuple, Optional
-
 
 class PriorityError(Exception):
     """
@@ -92,9 +89,9 @@ class Stream:
     def __init__(self, stream_id: int, weight: int = 16) -> None:
         self.stream_id = stream_id
         self.weight = weight
-        self.children: List[Stream] = []
-        self.parent: Optional[Stream] = None
-        self.child_queue: List[Tuple[int, Stream]] = []
+        self.children: list[Stream] = []
+        self.parent: Stream | None = None
+        self.child_queue: list[tuple[int, Stream]] = []
         self.active = True
         self.last_weight = 0
         self._deficit = 0
@@ -109,7 +106,7 @@ def weight(self, value: int) -> None:
         # weight between 1 and 256 (inclusive)."
         if not isinstance(value, int):
             raise BadWeightError("Stream weight should be an integer")
-        elif not (1 <= value <= 256):
+        if not (1 <= value <= 256):
             raise BadWeightError("Stream weight must be between 1 and 256 (inclusive)")
         self._weight = value
 
@@ -158,7 +155,7 @@ def remove_child(
         #   it in the old one
         self.children.remove(child)
 
-        new_queue: List[Tuple[int, Stream]] = []
+        new_queue: list[tuple[int, Stream]] = []
 
         while self.child_queue:
             level, stream = heapq.heappop(self.child_queue)
@@ -264,7 +261,7 @@ def _stream_cycle(new_parent: Stream, current: Stream) -> bool:
         parent = parent.parent  # type: ignore[assignment]
         if parent.stream_id == current.stream_id:
             return True
-        elif parent.stream_id == 0:
+        if parent.stream_id == 0:
             return False
 
     raise PriorityLoop(
@@ -339,7 +336,7 @@ def _exclusive_insert(
     def insert_stream(
         self,
         stream_id: int,
-        depends_on: Optional[int] = None,
+        depends_on: int | None = None,
         weight: int = 16,
         exclusive: bool = False,
     ) -> None:
@@ -383,7 +380,7 @@ def insert_stream(
     def reprioritize(
         self,
         stream_id: int,
-        depends_on: Optional[int] = None,
+        depends_on: int | None = None,
         weight: int = 16,
         exclusive: bool = False,
     ) -> None:

test/test_priority.py

@@ -1,26 +1,23 @@
-# -*- coding: utf-8 -*-
 """
 test_priority
 ~~~~~~~~~~~~~
 
 Tests for the Priority trees
 """
 
-import operator
 import collections
 import itertools
+import operator
+from collections.abc import Iterable
+from typing import Any
 
 import pytest
-
 from hypothesis import given, settings
-from hypothesis.stateful import invariant, RuleBasedStateMachine, rule
-from hypothesis.strategies import integers, lists, tuples, sampled_from
+from hypothesis.stateful import RuleBasedStateMachine, invariant, rule
+from hypothesis.strategies import integers, lists, sampled_from, tuples
 
 import priority
 
-from typing import Iterable, List, Dict, Any
-
-
 STREAMS_AND_WEIGHTS = lists(
     elements=tuples(integers(min_value=1), integers(min_value=1, max_value=255)),
     unique_by=operator.itemgetter(0),
@@ -59,7 +56,7 @@ def readme_tree():
 def active_readme_streams_from_filter(
     filtered: Iterable[int],
     blocked: bool = True,
-) -> List[int]:
+) -> list[int]:
     """
     Given a collection of filtered streams, determine which ones are active.
     This applies only to the readme tree at this time, though in future it
@@ -81,13 +78,13 @@ def active_readme_streams_from_filter(
     }
     filtered = set(filtered)
 
-    def get_expected(tree: Dict[Any, Any]) -> List[int]:
+    def get_expected(tree: dict[Any, Any]) -> list[int]:
         expected = []
 
         for stream_id in tree:
-            if stream_id not in filtered and blocked:
-                expected.append(stream_id)
-            elif stream_id in filtered and not blocked:
+            if (stream_id not in filtered and blocked) or (
+                stream_id in filtered and not blocked
+            ):
                 expected.append(stream_id)
             else:
                 expected.extend(get_expected(tree[stream_id]))
@@ -311,22 +308,22 @@ def test_priority_allows_inserting_stream_with_absent_parent(self, exclusive):
         p.insert_stream(stream_id=3, depends_on=1, exclusive=exclusive, weight=32)
 
         # Iterate 10 times to prove that the parent stream starts blocked.
-        first_ten_ids = [next(p) for _ in range(0, 10)]
+        first_ten_ids = [next(p) for _ in range(10)]
         assert first_ten_ids == [3] * 10
 
         # Unblock the parent.
         p.unblock(1)
 
         # Iterate 10 times, expecting only the parent.
-        next_ten_ids = [next(p) for _ in range(0, 10)]
+        next_ten_ids = [next(p) for _ in range(10)]
         assert next_ten_ids == [1] * 10
 
         # Insert a new stream into the tree with default priority.
         p.insert_stream(stream_id=5)
 
         # Iterate 10 more times. Expect the parent, and the new stream, in
         # equal amounts.
-        next_ten_ids = [next(p) for _ in range(0, 10)]
+        next_ten_ids = [next(p) for _ in range(10)]
         assert next_ten_ids == [5, 1] * 5
 
     @pytest.mark.parametrize("exclusive", [True, False])
@@ -341,22 +338,22 @@ def test_priority_reprioritizing_stream_with_absent_parent(self, exclusive):
         p.reprioritize(stream_id=3, depends_on=1, exclusive=exclusive, weight=32)
 
         # Iterate 10 times to prove that the parent stream starts blocked.
-        first_ten_ids = [next(p) for _ in range(0, 10)]
+        first_ten_ids = [next(p) for _ in range(10)]
         assert first_ten_ids == [3] * 10
 
         # Unblock the parent.
         p.unblock(1)
 
         # Iterate 10 times, expecting only the parent.
-        next_ten_ids = [next(p) for _ in range(0, 10)]
+        next_ten_ids = [next(p) for _ in range(10)]
         assert next_ten_ids == [1] * 10
 
         # Insert a new stream into the tree with default priority.
         p.insert_stream(stream_id=5)
 
         # Iterate 10 more times. Expect the parent, and the new stream, in
         # equal amounts.
-        next_ten_ids = [next(p) for _ in range(0, 10)]
+        next_ten_ids = [next(p) for _ in range(10)]
         assert next_ten_ids == [5, 1] * 5
 
     @pytest.mark.parametrize("count", range(2, 10000, 100))
@@ -386,7 +383,7 @@ def test_can_insert_stream_with_exclusive_dependency_on_0(self, depends_on):
 
         p.insert_stream(stream_id=5, depends_on=depends_on, exclusive=True)
 
-        next_ten_ids = [next(p) for _ in range(0, 10)]
+        next_ten_ids = [next(p) for _ in range(10)]
         assert next_ten_ids == [5] * 10
 
     @pytest.mark.parametrize("weight", [None, 0.5, float("inf"), "priority", object])