Enable pyupgrade

pyproject.toml

@@ -123,9 +123,8 @@ lint.select = [
   "F",   # pyflakes
   "I",   # isort
   "SIM", # flake8-simplify
-  # TODO: enable after dropping python 3.9 in a new PR.
-  # "UP",  # pyupgrade
-  "W", # pycodestyle warning
+  "UP",  # pyupgrade
+  "W",   # pycodestyle warning
 ]
 lint.ignore = [ "E501" ]
 

src/certwrangler/controllers.py

@@ -7,7 +7,6 @@
 
 import datetime
 import logging
-from typing import List, Optional, Tuple, Union
 
 import josepy as jose
 from acme import challenges as acme_challenges
@@ -86,7 +85,7 @@ def __init__(self, account: Account, state_manager: StateManager) -> None:
         self.state_manager = state_manager
         # Load to ensure we have the update state from the store.
         self.state_manager.load(self.account)
-        self._client: Optional[acme_client.ClientV2] = None
+        self._client: acme_client.ClientV2 | None = None
 
     @property
     def client(self) -> acme_client.ClientV2:
@@ -233,7 +232,7 @@ def __init__(self, cert: Cert, state_manager: StateManager) -> None:
         self.state_manager = state_manager
         # Load to ensure we have the update state from the store.
         self.state_manager.load(self.cert)
-        self._client: Optional[acme_client.ClientV2] = None
+        self._client: acme_client.ClientV2 | None = None
 
     @property
     def client(self) -> acme_client.ClientV2:
@@ -537,7 +536,7 @@ def _validate_authorizations(self) -> None:
 
     def _get_challenges(
         self, validate: bool = True, completed: bool = False
-    ) -> List[Tuple[str, acme_messages.ChallengeBody]]:
+    ) -> list[tuple[str, acme_messages.ChallengeBody]]:
         """
         Extracts the DNS challenges from the order.
 
@@ -570,7 +569,7 @@ def _get_challenges(
 
     def _get_dns_records(
         self, validate: bool = True, completed: bool = False
-    ) -> List[Tuple[str, str, str, Solver]]:
+    ) -> list[tuple[str, str, str, Solver]]:
         """
         Compiles and returns the parts of a DNS records and associated
         :class:`certwrangler.models.Solver` instances for each of the
@@ -611,7 +610,7 @@ def _get_dns_records(
             dns_records.append((name, zone, token, solver))
         return dns_records
 
-    def _fail_order(self, error: Optional[Union[Exception, str]] = None) -> None:
+    def _fail_order(self, error: Exception | str | None = None) -> None:
         """
         Cleans up any resources created as part of processing the order and
         removes the order from the cert's state.

src/certwrangler/daemon.py

@@ -4,9 +4,9 @@
 import signal
 import sys
 import threading
-from collections.abc import Iterable, Mapping
+from collections.abc import Callable, Iterable, Mapping
 from types import FrameType
-from typing import Any, Callable, List, Optional
+from typing import Any
 
 import click
 
@@ -35,17 +35,17 @@ class ThreadWithContext(threading.Thread):
     def __init__(
         self,
         group: None = None,
-        target: Optional[Callable[..., object]] = None,
-        name: Optional[str] = None,
+        target: Callable[..., object] | None = None,
+        name: str | None = None,
         args: Iterable[Any] = [],
-        kwargs: Optional[Mapping[str, Any]] = None,
+        kwargs: Mapping[str, Any] | None = None,
         *,
-        daemon: Optional[bool] = None,
+        daemon: bool | None = None,
     ) -> None:
         self.ctx: click.Context = click.get_current_context()
-        self.loop: Optional[asyncio.AbstractEventLoop] = None
+        self.loop: asyncio.AbstractEventLoop | None = None
         self.graceful_stop_event: threading.Event = threading.Event()
-        self.async_graceful_stop_event: Optional[asyncio.Event] = None
+        self.async_graceful_stop_event: asyncio.Event | None = None
         super().__init__(group, target, name, args, kwargs, daemon=daemon)
 
     def run(self) -> None:
@@ -121,7 +121,7 @@ class Daemon:
 
     def __init__(self) -> None:
         self.ctx: click.Context = click.get_current_context()
-        self.threads: List[ThreadWithContext] = []
+        self.threads: list[ThreadWithContext] = []
         self.stopping: threading.Event = threading.Event()
         self.reloading: threading.Lock = threading.Lock()
 
@@ -161,7 +161,7 @@ def _stop_threads(self) -> None:
             log.info(f"Thread {thread.name} stopped.")
         self.threads = [thread for thread in self.threads if thread.is_alive()]
 
-    def _reload_handler(self, signal_number: int, _frame: Optional[FrameType]) -> None:
+    def _reload_handler(self, signal_number: int, _frame: FrameType | None) -> None:
         """
         Handles processing a reload of the config on SIGHUP. This will gracefully
         stop the threads, reload and re-initialize the config, then start the
@@ -178,7 +178,7 @@ def _reload_handler(self, signal_number: int, _frame: Optional[FrameType]) -> No
             self._create_threads()
             self._start_threads()
 
-    def _stop_handler(self, signal_number: int, _frame: Optional[FrameType]) -> None:
+    def _stop_handler(self, signal_number: int, _frame: FrameType | None) -> None:
         """
         Handles processing stopping the threads on SIGTERM and SIGINT. This
         will first attempt to gracefully stop the threads, then forcefully

src/certwrangler/dns.py

@@ -1,7 +1,6 @@
 import logging
 import time
 from datetime import datetime, timedelta
-from typing import List, Tuple
 
 import click
 from dns.rdatatype import RdataType
@@ -13,7 +12,7 @@
 @click.pass_context
 def wait_for_challenges(
     ctx: click.Context,
-    dns_records: List[Tuple[str, str]],
+    dns_records: list[tuple[str, str]],
     wait_timeout: timedelta,
     sleep: int = 5,
 ) -> None:

src/certwrangler/metrics.py

@@ -3,10 +3,9 @@
 import logging
 import types
 from collections import UserDict
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from dataclasses import dataclass, field
 from functools import partial
-from typing import Callable, Dict, List, Optional, Union
 
 import click
 from prometheus_client import Counter, Gauge, Info
@@ -15,8 +14,7 @@
 log = logging.getLogger(__name__)
 
 GaugeFunction = Callable[[click.Context, str], float]
-InfoFunction = Callable[[click.Context, str], Dict[str, str]]
-MetricTypes = Union[Counter, Gauge]
+InfoFunction = Callable[[click.Context, str], dict[str, str]]
 
 
 class DynamicInfo(Info):
@@ -25,7 +23,7 @@ class DynamicInfo(Info):
     pulling samples from a function, similar to gauges.
     """
 
-    def set_function(self, f: Callable[[], Dict[str, str]]) -> None:
+    def set_function(self, f: Callable[[], dict[str, str]]) -> None:
         """Call the provided function to return the Info value.
 
         The function must return a dict of strings.
@@ -55,9 +53,9 @@ class EntityMetrics:
     A registry of labeled metrics for an entity.
     """
 
-    counters: Dict[str, Counter] = field(default_factory=dict)
-    gauges: Dict[str, Gauge] = field(default_factory=dict)
-    infos: Dict[str, DynamicInfo] = field(default_factory=dict)
+    counters: dict[str, Counter] = field(default_factory=dict)
+    gauges: dict[str, Gauge] = field(default_factory=dict)
+    infos: dict[str, DynamicInfo] = field(default_factory=dict)
 
 
 class MetricRegistry(UserDict[str, EntityMetrics]):
@@ -70,12 +68,12 @@ class MetricRegistry(UserDict[str, EntityMetrics]):
 
     def __init__(self, subsystem: str) -> None:
         self.subsystem = subsystem
-        self.data: Dict[str, EntityMetrics] = {}
-        self._counters: Dict[str, Counter] = {}
-        self._gauges: Dict[str, Gauge] = {}
-        self._gauge_functions: Dict[str, Optional[GaugeFunction]] = {}
-        self._infos: Dict[str, DynamicInfo] = {}
-        self._info_functions: Dict[str, Optional[InfoFunction]] = {}
+        self.data: dict[str, EntityMetrics] = {}
+        self._counters: dict[str, Counter] = {}
+        self._gauges: dict[str, Gauge] = {}
+        self._gauge_functions: dict[str, GaugeFunction | None] = {}
+        self._infos: dict[str, DynamicInfo] = {}
+        self._info_functions: dict[str, InfoFunction | None] = {}
 
     def add_counter(
         self,
@@ -108,7 +106,7 @@ def add_gauge(
         name: str,
         documentation: str,
         unit: str = "",
-        function: Optional[GaugeFunction] = None,
+        function: GaugeFunction | None = None,
     ) -> None:
         """
         Register a new gauge in the registry.
@@ -136,7 +134,7 @@ def add_info(
         self,
         name: str,
         documentation: str,
-        function: Optional[InfoFunction] = None,
+        function: InfoFunction | None = None,
     ) -> None:
         """
         Register a new info in the registry.
@@ -158,7 +156,7 @@ def add_info(
         )
         self._info_functions[name] = function
 
-    def reconcile_entities(self, entities: List[str]) -> None:
+    def reconcile_entities(self, entities: list[str]) -> None:
         """
         Takes the list of entities that should be present and updates the
         registry to reflect that.