From 5a94709a9cffefb9908331e061033ceeea687422 Mon Sep 17 00:00:00 2001
From: "M. David Bennett" <20524377+mdavidbennett@users.noreply.github.com>
Date: Mon, 15 Jun 2026 09:52:48 -0500
Subject: [PATCH] Enable pyupgrade

This enabled pyupgrade and applies all the changes it suggested. I also
removed the unused `MetricTypes` type from `metrics.py`.
---
 pyproject.toml                           |   5 +-
 src/certwrangler/controllers.py          |  11 ++-
 src/certwrangler/daemon.py               |  22 ++---
 src/certwrangler/dns.py                  |   3 +-
 src/certwrangler/metrics.py              |  32 ++++---
 src/certwrangler/models.py               | 104 +++++++++++------------
 src/certwrangler/schema_migrations.py    |   9 +-
 src/certwrangler/shell.py                |  10 +--
 src/certwrangler/solvers/edgedns.py      |  10 +--
 src/certwrangler/solvers/lexicon.py      |   6 +-
 src/certwrangler/state_managers/dummy.py |  10 +--
 src/certwrangler/state_managers/local.py |  18 ++--
 src/certwrangler/stores/local.py         |   4 +-
 src/certwrangler/stores/vault.py         |  20 ++---
 src/certwrangler/types.py                |   4 +-
 src/certwrangler/utils.py                |   5 +-
 16 files changed, 133 insertions(+), 140 deletions(-)

diff --git a/pyproject.toml b/pyproject.toml
index 0b71656..3b529c0 100644
--- a/pyproject.toml
+++ b/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" ]
 
diff --git a/src/certwrangler/controllers.py b/src/certwrangler/controllers.py
index eb1ff3c..9db9d39 100644
--- a/src/certwrangler/controllers.py
+++ b/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.
diff --git a/src/certwrangler/daemon.py b/src/certwrangler/daemon.py
index 71e2179..e39460c 100644
--- a/src/certwrangler/daemon.py
+++ b/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
diff --git a/src/certwrangler/dns.py b/src/certwrangler/dns.py
index ceec547..14db69c 100644
--- a/src/certwrangler/dns.py
+++ b/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:
diff --git a/src/certwrangler/metrics.py b/src/certwrangler/metrics.py
index 8fb94db..7ebe48c 100644
--- a/src/certwrangler/metrics.py
+++ b/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.
diff --git a/src/certwrangler/models.py b/src/certwrangler/models.py
index b02f74b..f7f03f4 100644
--- a/src/certwrangler/models.py
+++ b/src/certwrangler/models.py
@@ -8,13 +8,15 @@
 
 import abc
 import base64
+import builtins
 import hashlib
 import logging
+from collections.abc import Callable
 from datetime import datetime, timedelta, timezone
 from enum import Enum
 from ipaddress import IPv4Address
 from pathlib import Path
-from typing import Any, Callable, ClassVar, Dict, List, Literal, Optional, Union
+from typing import Any, ClassVar, Literal
 
 from cryptography import x509
 from cryptography.fernet import MultiFernet
@@ -83,7 +85,7 @@ class StateModel(BaseModel):
     The :attr:`_migrated` attribute is set if the model schema was migrated.
     """
 
-    schema_migrations: ClassVar[List[Callable[[Dict[str, Any]], Dict[str, Any]]]] = []
+    schema_migrations: ClassVar[list[Callable[[dict[str, Any]], dict[str, Any]]]] = []
 
     _migrated: bool = PrivateAttr(default=False)
 
@@ -132,7 +134,7 @@ class Solver(NamedModel, metaclass=abc.ABCMeta):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     driver: str = Field(..., description="The name of the driver to use.")
-    zones: List[Domain] = Field(
+    zones: list[Domain] = Field(
         ..., description="A list of DNS zones this solver should be used for."
     )
 
@@ -159,7 +161,7 @@ def initialize(self) -> None:
 
     @field_validator("zones")
     @classmethod
-    def __validate_zones(cls, values: List[Domain]) -> List[Domain]:
+    def __validate_zones(cls, values: list[Domain]) -> list[Domain]:
         """
         Validate that the configured zones have valid SOA records.
 
@@ -204,7 +206,7 @@ class StateManager(BaseModel, metaclass=abc.ABCMeta):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     driver: str = Field(..., description="The name of the driver to use.")
-    encryption_keys: List[FernetKey] = Field(
+    encryption_keys: builtins.list[FernetKey] = Field(
         default_factory=list,
         description="An optional list of encryption keys to use to encrypt "
         "the state. Only the top-most key will be used for encryption "
@@ -214,10 +216,10 @@ class StateManager(BaseModel, metaclass=abc.ABCMeta):
     )
 
     _config: Config = PrivateAttr()
-    _encryptor: Optional[Encryptor] = PrivateAttr(default=None)
+    _encryptor: Encryptor | None = PrivateAttr(default=None)
 
     @property
-    def encryptor(self) -> Optional[Encryptor]:
+    def encryptor(self) -> Encryptor | None:
         """
         This sets up and returns an Encryptor if ``encryption_keys`` are defined.
 
@@ -236,21 +238,21 @@ def initialize(self) -> None:
         pass
 
     @abc.abstractmethod
-    def list(self) -> Dict[str, Dict[str, Any]]:
+    def list(self) -> dict[str, dict[str, Any]]:
         """
         Lists all the saved states for the given entity_class including encryption fingerprint.
         """
         raise NotImplementedError
 
     @abc.abstractmethod
-    def save(self, entity: Union[Account, Cert], encrypt: bool = True) -> None:
+    def save(self, entity: Account | Cert, encrypt: bool = True) -> None:
         """
         Saves the state of the given entity.
         """
         raise NotImplementedError
 
     @abc.abstractmethod
-    def load(self, entity: Union[Account, Cert]) -> None:
+    def load(self, entity: Account | Cert) -> None:
         """
         Loads the state of the given entity to memory.
         """
@@ -258,7 +260,7 @@ def load(self, entity: Union[Account, Cert]) -> None:
 
     @abc.abstractmethod
     def delete(
-        self, entity_class: Union[Literal["account"], Literal["cert"]], entity_name: str
+        self, entity_class: Literal["account"] | Literal["cert"], entity_name: str
     ) -> None:
         """
         Deletes the given entity_name from state.
@@ -310,11 +312,11 @@ class AccountState(StateModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
     schema_migrations = ACCOUNT_STATE_SCHEMA_MIGRATIONS
 
-    registration: Optional[Registration] = Field(
+    registration: Registration | None = Field(
         None, description="The ACME registration record."
     )
-    key: Optional[JWKRSAKey] = Field(None, description="The current RSA key.")
-    key_size: Optional[int] = Field(
+    key: JWKRSAKey | None = Field(None, description="The current RSA key.")
+    key_size: int | None = Field(
         None, description="The size of the current RSA key in bits."
     )
     status: AccountStatus = AccountStatus.new
@@ -327,7 +329,7 @@ class Account(NamedModel):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    emails: List[EmailStr] = Field(
+    emails: list[EmailStr] = Field(
         ..., description="A list of email addresses for the account."
     )
     server: HttpUrl = Field(
@@ -348,7 +350,7 @@ def state(self, value: AccountState) -> None:
 
     @field_validator("emails")
     @classmethod
-    def __validate_unique_emails(cls, values: List[EmailStr]) -> List[EmailStr]:
+    def __validate_unique_emails(cls, values: list[EmailStr]) -> list[EmailStr]:
         """
         Validates that all the configured emails are unique.
         """
@@ -364,15 +366,15 @@ class Subject(NamedModel):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    country: Optional[CountryNameOID] = Field(None, description="The country name OID.")  # type: ignore
-    state_or_province: Optional[StateOrProvinceOID] = Field(  # type: ignore
+    country: CountryNameOID | None = Field(None, description="The country name OID.")  # type: ignore
+    state_or_province: StateOrProvinceOID | None = Field(  # type: ignore
         None, description="The state or province OID."
     )
-    locality: Optional[LocalityOID] = Field(None, description="The locality OID.")  # type: ignore
-    organization: Optional[OrganizationOID] = Field(  # type: ignore
+    locality: LocalityOID | None = Field(None, description="The locality OID.")  # type: ignore
+    organization: OrganizationOID | None = Field(  # type: ignore
         None, description="The organization OID."
     )
-    organizational_unit: Optional[OrganizationalUnitOID] = Field(  # type: ignore
+    organizational_unit: OrganizationalUnitOID | None = Field(  # type: ignore
         None, description="The organizational unit OID."
     )
 
@@ -391,29 +393,27 @@ class CertState(StateModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
     schema_migrations = CERT_STATE_SCHEMA_MIGRATIONS
 
-    url: Optional[str] = Field(
+    url: str | None = Field(
         None, description="The URL of the cert retrieved from the ACME server."
     )
-    key: Optional[RSAKey] = Field(None, description="The cert's RSA key.")
-    key_size: Optional[int] = Field(
-        None, description="The size of the RSA key in bits."
-    )
-    cert: Optional[X509Certificate] = Field(
+    key: RSAKey | None = Field(None, description="The cert's RSA key.")
+    key_size: int | None = Field(None, description="The size of the RSA key in bits.")
+    cert: X509Certificate | None = Field(
         None, description="The cert returned by the ACME server."
     )
-    chain: Optional[List[X509Certificate]] = Field(
+    chain: list[X509Certificate] | None = Field(
         None, description="The chain of trust returned by the ACME server."
     )
-    csr: Optional[X509CSR] = Field(
+    csr: X509CSR | None = Field(
         None, description="The CSR generated to request the cert."
     )
-    order: Optional[Order] = Field(
+    order: Order | None = Field(
         None, description="The order if an order is currently active."
     )
     status: CertStatus = CertStatus.new
 
     @computed_field
-    def fullchain(self) -> Optional[List[X509Certificate]]:
+    def fullchain(self) -> list[X509Certificate] | None:
         """
         The full chain of trust including the leaf cert.
         """
@@ -437,11 +437,11 @@ class Cert(NamedModel):
         description="The name of the configured ACME account the cert should "
         "be created under.",
     )
-    store_names: List[str] = Field(
+    store_names: list[str] = Field(
         ...,
         description="A list of the configured stores the cert should be published to.",
     )
-    store_key: Optional[str] = Field(
+    store_key: str | None = Field(
         None,
         description="Optional sub-key the cert should be published to in the "
         "store. Currently only supported by the vault store driver.",
@@ -451,7 +451,7 @@ class Cert(NamedModel):
         description="The name of the configured subject the cert should be "
         "created with.",
     )
-    alt_names: List[Domain] = Field(
+    alt_names: list[Domain] = Field(
         default_factory=list, description="A list of alternative names for the cert."
     )
     wait_timeout: timedelta = Field(
@@ -490,7 +490,7 @@ def account(self) -> Account:
             raise ValueError(f"No account named '{self.account_name}'.") from error
 
     @property
-    def stores(self) -> List[Store]:
+    def stores(self) -> list[Store]:
         """
         Returns a list of the configured store objects.
 
@@ -508,7 +508,7 @@ def stores(self) -> List[Store]:
         return stores
 
     @property
-    def solvers(self) -> Dict[str, Solver]:
+    def solvers(self) -> dict[str, Solver]:
         """
         Returns the available solvers.
         """
@@ -540,7 +540,7 @@ def get_solver_for_zone(self, zone: str) -> Solver:
 
     @field_validator("store_names")
     @classmethod
-    def __validate_unique_stores(cls, values: List[str]) -> List[str]:
+    def __validate_unique_stores(cls, values: list[str]) -> list[str]:
         """
         Validates that all the configured stores are unique.
 
@@ -637,16 +637,14 @@ class HttpConfig(BaseModel):
     )
     port: int = Field(6377, description="Port the HTTP server should listen on.")
     server_name: str = Field("certwrangler", description="Name of the HTTP server.")
-    ssl_key_file: Optional[Path] = Field(None, description="Optional SSL key.")
-    ssl_key_password: Optional[str] = Field(
-        None, description="Optional SSL key password."
-    )
-    ssl_cert_file: Optional[Path] = Field(None, description="Optional SSL cert.")
-    ssl_ca_certs_file: Optional[Path] = Field(None, description="Optional SSL CA cert.")
+    ssl_key_file: Path | None = Field(None, description="Optional SSL key.")
+    ssl_key_password: str | None = Field(None, description="Optional SSL key password.")
+    ssl_cert_file: Path | None = Field(None, description="Optional SSL cert.")
+    ssl_ca_certs_file: Path | None = Field(None, description="Optional SSL CA cert.")
 
     @field_validator("ssl_key_file", "ssl_cert_file", "ssl_ca_certs_file")
     @classmethod
-    def __validate_ssl_files_exist(cls, value: Optional[Path]) -> Optional[Path]:
+    def __validate_ssl_files_exist(cls, value: Path | None) -> Path | None:
         """
         Validates that the specified file exists.
 
@@ -714,15 +712,15 @@ class Config(BaseModel):
     state_manager: StateManager = Field(
         ..., alias="state", description="Config for the state manager."
     )
-    accounts: Dict[str, Account] = Field(..., description="Config for the accounts.")
-    certs: Dict[str, Cert] = Field(..., description="Config for the certs.")
-    solvers: Dict[str, Solver] = Field(..., description="Config for the solvers.")
-    stores: Dict[str, Store] = Field(..., description="Config for the stores.")
-    subjects: Dict[str, Subject] = Field(..., description="Config for the subjects.")
+    accounts: dict[str, Account] = Field(..., description="Config for the accounts.")
+    certs: dict[str, Cert] = Field(..., description="Config for the certs.")
+    solvers: dict[str, Solver] = Field(..., description="Config for the solvers.")
+    stores: dict[str, Store] = Field(..., description="Config for the stores.")
+    subjects: dict[str, Subject] = Field(..., description="Config for the subjects.")
 
     @field_validator("solvers", mode="before")
     @classmethod
-    def __load_solver_plugins(cls, values: Dict[str, Any]) -> Dict[str, Solver]:
+    def __load_solver_plugins(cls, values: dict[str, Any]) -> dict[str, Solver]:
         """
         Dynamically load solver plugins based on their driver key.
 
@@ -742,7 +740,7 @@ def __load_solver_plugins(cls, values: Dict[str, Any]) -> Dict[str, Solver]:
 
     @field_validator("state_manager", mode="before")
     @classmethod
-    def __load_state_manager_plugin(cls, values: Dict[str, Any]) -> StateManager:
+    def __load_state_manager_plugin(cls, values: dict[str, Any]) -> StateManager:
         """
         Dynamically load state_manager plugins based on their driver key.
 
@@ -760,7 +758,7 @@ def __load_state_manager_plugin(cls, values: Dict[str, Any]) -> StateManager:
 
     @field_validator("stores", mode="before")
     @classmethod
-    def __load_store_plugins(cls, values: Dict[str, Any]) -> Dict[str, Store]:
+    def __load_store_plugins(cls, values: dict[str, Any]) -> dict[str, Store]:
         """
         Dynamically load store plugins based on their driver key.
 
@@ -780,7 +778,7 @@ def __load_store_plugins(cls, values: Dict[str, Any]) -> Dict[str, Store]:
 
     @model_validator(mode="before")
     @classmethod
-    def __pre_populate(cls, values: Dict[str, Any]) -> Dict[str, Any]:
+    def __pre_populate(cls, values: dict[str, Any]) -> dict[str, Any]:
         """
         Pre-populate the config data with some defaults.
         """
diff --git a/src/certwrangler/schema_migrations.py b/src/certwrangler/schema_migrations.py
index b62e1ea..f67c8cf 100644
--- a/src/certwrangler/schema_migrations.py
+++ b/src/certwrangler/schema_migrations.py
@@ -2,10 +2,11 @@
 This module contains state schema migrations that should be applied.
 """
 
-from typing import Any, Callable, Dict, List
+from collections.abc import Callable
+from typing import Any
 
 
-def _cert_migration_00_add_chain(data: Dict[str, Any]) -> Dict[str, Any]:
+def _cert_migration_00_add_chain(data: dict[str, Any]) -> dict[str, Any]:
     """
     Switch from storing the CA and intermediate(s) separately to a generic
     chain field.
@@ -27,7 +28,7 @@ def _cert_migration_00_add_chain(data: Dict[str, Any]) -> Dict[str, Any]:
     return data
 
 
-ACCOUNT_STATE_SCHEMA_MIGRATIONS: List[Callable[[Dict[str, Any]], Dict[str, Any]]] = []
-CERT_STATE_SCHEMA_MIGRATIONS: List[Callable[[Dict[str, Any]], Dict[str, Any]]] = [
+ACCOUNT_STATE_SCHEMA_MIGRATIONS: list[Callable[[dict[str, Any]], dict[str, Any]]] = []
+CERT_STATE_SCHEMA_MIGRATIONS: list[Callable[[dict[str, Any]], dict[str, Any]]] = [
     _cert_migration_00_add_chain,
 ]
diff --git a/src/certwrangler/shell.py b/src/certwrangler/shell.py
index 038e1b8..5223c26 100644
--- a/src/certwrangler/shell.py
+++ b/src/certwrangler/shell.py
@@ -5,7 +5,7 @@
 import sys
 from importlib.util import find_spec
 from ipaddress import ip_address
-from typing import Any, Dict, List, Tuple
+from typing import Any
 
 import click
 from cryptography.fernet import Fernet
@@ -18,8 +18,8 @@
 
 
 def _validate_nameservers(
-    ctx: click.Context, params: Dict[str, Any], values: List[str]
-) -> List[str]:
+    ctx: click.Context, params: dict[str, Any], values: list[str]
+) -> list[str]:
     for value in values:
         try:
             ip_address(value)
@@ -66,7 +66,7 @@ def _validate_nameservers(
 )
 @click.pass_context
 def cli(
-    ctx: click.Context, config: str, log_level: str, nameservers: List[str]
+    ctx: click.Context, config: str, log_level: str, nameservers: list[str]
 ) -> None:
     """The certwrangler management cli."""
 
@@ -132,7 +132,7 @@ def run(ctx: click.Context) -> None:
     @cli.command(context_settings={"ignore_unknown_options": True})
     @click.argument("ipython_args", nargs=-1, type=click.UNPROCESSED)
     @click.pass_context
-    def dev_shell(ctx: click.Context, ipython_args: Tuple[Any]) -> None:
+    def dev_shell(ctx: click.Context, ipython_args: tuple[Any]) -> None:
         """Open an IPython shell with a certwrangler context."""
 
         ctx.obj.load_config()
diff --git a/src/certwrangler/solvers/edgedns.py b/src/certwrangler/solvers/edgedns.py
index e867930..d500348 100644
--- a/src/certwrangler/solvers/edgedns.py
+++ b/src/certwrangler/solvers/edgedns.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Dict, Literal, Optional
+from typing import Any, Literal
 
 import requests
 
@@ -119,7 +119,7 @@ def delete(self, name: str, domain: str, content: str) -> None:
                 f"Expected 'rdata' in response to be a list, got {type(record.get('rdata')).__name__}."
             )
 
-    def _cleanup_response(self, response: Dict[str, Any]) -> Dict[str, Any]:
+    def _cleanup_response(self, response: dict[str, Any]) -> dict[str, Any]:
         """
         The EdgeDNS API seems to leave literal double quotes on the strings for
         TXT records. This just goes through each of the entries to strip them away.
@@ -153,7 +153,7 @@ def _delete(self, endpoint: str) -> bytes:
             raise SolverError(error) from error
         return response.content
 
-    def _get(self, endpoint: str) -> Optional[Dict[str, Any]]:
+    def _get(self, endpoint: str) -> dict[str, Any] | None:
         """
         Send a GET request to the specified endpoint and return the parsed
         json response.
@@ -173,7 +173,7 @@ def _get(self, endpoint: str) -> Optional[Dict[str, Any]]:
             raise SolverError(error) from error
         return self._cleanup_response(response.json())
 
-    def _post(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+    def _post(self, endpoint: str, payload: dict[str, Any]) -> dict[str, Any]:
         """
         Send a POST request to the specified endpoint with the specified payload
         and return the parsed json response.
@@ -191,7 +191,7 @@ def _post(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
             raise SolverError(error) from error
         return self._cleanup_response(response.json())
 
-    def _put(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+    def _put(self, endpoint: str, payload: dict[str, Any]) -> dict[str, Any]:
         """
         Send a PUT request to the specified endpoint with the specified payload
         and return the parsed json response.
diff --git a/src/certwrangler/solvers/lexicon.py b/src/certwrangler/solvers/lexicon.py
index 152e817..23cac7d 100644
--- a/src/certwrangler/solvers/lexicon.py
+++ b/src/certwrangler/solvers/lexicon.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Dict, Literal
+from typing import Any, Literal
 
 from lexicon.client import Client
 from lexicon.config import ConfigResolver
@@ -27,7 +27,7 @@ class LexiconSolver(Solver):
     provider_name: str = Field(
         ..., description="The name of the lexicon provider to use."
     )
-    provider_options: Dict[str, Any] = Field(
+    provider_options: dict[str, Any] = Field(
         default_factory=dict, description="Provider-specific options."
     )
 
@@ -65,7 +65,7 @@ def delete(self, name: str, domain: str, content: str) -> None:
 
     def _build_config(
         self, action: str, name: str, domain: str, content: str
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Generate the needed lexicon config for the request based on the action
         and the provider_options.
diff --git a/src/certwrangler/state_managers/dummy.py b/src/certwrangler/state_managers/dummy.py
index 83c80ee..5449a89 100644
--- a/src/certwrangler/state_managers/dummy.py
+++ b/src/certwrangler/state_managers/dummy.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Dict, Literal, Union
+from typing import Any, Literal
 
 from certwrangler.models import Account, Cert, StateManager
 
@@ -21,14 +21,14 @@ def initialize(self) -> None:
         """
         log.info("initialize called on dummy state manager")
 
-    def list(self) -> Dict[str, Dict[str, Any]]:
+    def list(self) -> dict[str, dict[str, Any]]:
         """
         No-op list, just log we were here and return an empty dict.
         """
         log.info("list called on dummy state manager")
         return {}
 
-    def save(self, entity: Union[Account, Cert], encrypt: bool = True) -> None:
+    def save(self, entity: Account | Cert, encrypt: bool = True) -> None:
         """
         No-op save, just log we were here.
         """
@@ -37,7 +37,7 @@ def save(self, entity: Union[Account, Cert], encrypt: bool = True) -> None:
             f"save called with {entity_type} '{entity.name}' encrypt={encrypt} on dummy state manager"
         )
 
-    def load(self, entity: Union[Account, Cert]) -> None:
+    def load(self, entity: Account | Cert) -> None:
         """
         No-op load, just log we were here.
         """
@@ -47,7 +47,7 @@ def load(self, entity: Union[Account, Cert]) -> None:
         )
 
     def delete(
-        self, entity_class: Union[Literal["account"], Literal["cert"]], entity_name: str
+        self, entity_class: Literal["account"] | Literal["cert"], entity_name: str
     ) -> None:
         """
         No-op delete, just log we were here.
diff --git a/src/certwrangler/state_managers/local.py b/src/certwrangler/state_managers/local.py
index 20125d9..4b17914 100644
--- a/src/certwrangler/state_managers/local.py
+++ b/src/certwrangler/state_managers/local.py
@@ -5,7 +5,7 @@
 import re
 import textwrap
 from pathlib import Path
-from typing import Any, Dict, List, Literal, Optional, Union
+from typing import Any, Literal
 
 from cryptography.fernet import InvalidToken
 from pydantic import Field
@@ -44,7 +44,7 @@ def _is_encrypted(data: str) -> bool:
     return bool(re.match(ENCRYPTION_REGEX, data))
 
 
-def _parse_encrypted_state(data: str) -> Dict[str, Any]:
+def _parse_encrypted_state(data: str) -> dict[str, Any]:
     """
     Parses the encrypted state and returns a dict with the encrypted data
     and any discovered metadata tags.
@@ -79,7 +79,7 @@ def _decrypt(data: str, encryptor: Encryptor) -> str:
 
 
 def _encrypt(
-    encryptor: Encryptor, data: str, metadata: Optional[Dict[str, str]] = None
+    encryptor: Encryptor, data: str, metadata: dict[str, str] | None = None
 ) -> str:
     """
     Encrypts the contents of data using the provided encryptor. Can optionally
@@ -106,8 +106,8 @@ def _encrypt(
 
 
 def _list_entities(
-    state_path_dir: Path, known_entities: List[str]
-) -> Dict[str, Dict[str, Any]]:
+    state_path_dir: Path, known_entities: list[str]
+) -> dict[str, dict[str, Any]]:
     """
     Loops through the contents of the provided state_path_dir and compares
     the discovered entities to the provided known_entities list. Returns
@@ -186,7 +186,7 @@ def initialize(self) -> None:
         except OSError as error:
             raise StateManagerError(error) from error
 
-    def list(self) -> Dict[str, Dict[str, Any]]:
+    def list(self) -> dict[str, dict[str, Any]]:
         """
         List all the state entities under management. Returns a dict of all
         the names of accounts and certs it discovers.
@@ -205,7 +205,7 @@ def list(self) -> Dict[str, Dict[str, Any]]:
         except OSError as error:
             raise StateManagerError(error) from error
 
-    def save(self, entity: Union[Account, Cert], encrypt: bool = True) -> None:
+    def save(self, entity: Account | Cert, encrypt: bool = True) -> None:
         """
         Save the provided entity's (Account or Cert object) state and by
         default will encrypt the contents if an encryptor is configured.
@@ -232,7 +232,7 @@ def save(self, entity: Union[Account, Cert], encrypt: bool = True) -> None:
         except OSError as error:
             raise StateManagerError(error) from error
 
-    def load(self, entity: Union[Account, Cert]) -> None:
+    def load(self, entity: Account | Cert) -> None:
         """
         Load and decrypt (if an encryptor is present) the state of the provided
         entity (Account or Cert object).
@@ -269,7 +269,7 @@ def load(self, entity: Union[Account, Cert]) -> None:
             ) from error
 
     def delete(
-        self, entity_class: Union[Literal["account"], Literal["cert"]], entity_name: str
+        self, entity_class: Literal["account"] | Literal["cert"], entity_name: str
     ) -> None:
         """
         Delete the state for the provided entity_class and entity_name.
diff --git a/src/certwrangler/stores/local.py b/src/certwrangler/stores/local.py
index ee4292c..23be747 100644
--- a/src/certwrangler/stores/local.py
+++ b/src/certwrangler/stores/local.py
@@ -3,7 +3,7 @@
 import hashlib
 import logging
 from pathlib import Path
-from typing import Literal, Union
+from typing import Literal
 
 from pydantic import Field
 
@@ -89,7 +89,7 @@ def publish(self, cert: Cert) -> None:
             except OSError as error:
                 raise StoreError(error) from error
 
-    def _get_digest(self, obj: Union[str, Path]) -> str:
+    def _get_digest(self, obj: str | Path) -> str:
         """
         Get the sha256sum of either a str or Path object.
         """
diff --git a/src/certwrangler/stores/vault.py b/src/certwrangler/stores/vault.py
index 423fb9f..2b4d30a 100644
--- a/src/certwrangler/stores/vault.py
+++ b/src/certwrangler/stores/vault.py
@@ -3,7 +3,7 @@
 import abc
 import logging
 from pathlib import Path
-from typing import Dict, Literal, Optional, Union
+from typing import Literal
 
 # hvac has no types, see: https://github.com/hvac/hvac/issues/800
 import hvac  # type: ignore
@@ -37,7 +37,7 @@ class AppRoleAuth(BaseAuth):
     """
 
     method: Literal["approle"]
-    mount_point: Optional[str] = Field(
+    mount_point: str | None = Field(
         default=None, description="Optional mount point for the auth method."
     )
     role_id: str = Field(..., description="The AppRole role_id.")
@@ -63,7 +63,7 @@ class KubernetesAuth(BaseAuth):
     """
 
     method: Literal["kubernetes"]
-    mount_point: Optional[str] = Field(
+    mount_point: str | None = Field(
         default=None, description="Optional mount point for the auth method."
     )
     role: str = Field(..., description="The name of the role.")
@@ -111,7 +111,7 @@ class VaultStore(Store):
 
     driver: Literal["vault"]
     server: HttpUrl = Field(..., description="The URI of the vault server.")
-    ca_cert: Optional[Path] = Field(
+    ca_cert: Path | None = Field(
         default=None,
         description="Optional path to a CA cert for requests to vault.",
     )
@@ -120,11 +120,11 @@ class VaultStore(Store):
     version: Literal[1, 2] = Field(
         default=2, description="The version of the vault secrets engine."
     )
-    auth: Union[AppRoleAuth, TokenAuth, KubernetesAuth] = Field(
+    auth: AppRoleAuth | TokenAuth | KubernetesAuth = Field(
         discriminator="method", description="The config for authenticating with vault."
     )
 
-    _client: Optional[hvac.Client] = PrivateAttr(default=None)
+    _client: hvac.Client | None = PrivateAttr(default=None)
 
     @property
     def client(self) -> hvac.Client:
@@ -190,7 +190,7 @@ def publish(self, cert: Cert) -> None:
                 f"Cert '{cert.name}' published to '{self.mount_point}/{secret_path}' in vault"
             )
 
-    def _read_v1(self, path: Path) -> Dict[str, str]:
+    def _read_v1(self, path: Path) -> dict[str, str]:
         """
         Read the contents of a secret from a v1 vault endpoint.
         """
@@ -200,7 +200,7 @@ def _read_v1(self, path: Path) -> Dict[str, str]:
             path=path,
         )["data"]
 
-    def _write_v1(self, path: Path, secret: Dict[str, str]) -> None:
+    def _write_v1(self, path: Path, secret: dict[str, str]) -> None:
         """
         Write the contents of a secret to a v1 vault endpoint.
         """
@@ -211,7 +211,7 @@ def _write_v1(self, path: Path, secret: Dict[str, str]) -> None:
             secret=secret,
         )
 
-    def _read_v2(self, path: Path) -> Dict[str, str]:
+    def _read_v2(self, path: Path) -> dict[str, str]:
         """
         Read the contents of a secret from a v2 vault endpoint.
         """
@@ -221,7 +221,7 @@ def _read_v2(self, path: Path) -> Dict[str, str]:
             path=path,
         )["data"]["data"]
 
-    def _write_v2(self, path: Path, secret: Dict[str, str]) -> None:
+    def _write_v2(self, path: Path, secret: dict[str, str]) -> None:
         """
         Write the contents of a secret to a v1 vault endpoint.
         """
diff --git a/src/certwrangler/types.py b/src/certwrangler/types.py
index 756ca88..c6a930e 100644
--- a/src/certwrangler/types.py
+++ b/src/certwrangler/types.py
@@ -1,5 +1,5 @@
 from datetime import timedelta
-from typing import Annotated, Any, Dict, Union
+from typing import Annotated, Any
 
 import josepy as jose
 from acme import messages
@@ -174,7 +174,7 @@
 
 
 def _order_loader(
-    value: Union[Dict[str, Any], messages.OrderResource],
+    value: dict[str, Any] | messages.OrderResource,
 ) -> messages.OrderResource:
     """
     Deserializes a json representation of :class:`acme.messages.OrderResource`,
diff --git a/src/certwrangler/utils.py b/src/certwrangler/utils.py
index e76ba19..7757c1d 100644
--- a/src/certwrangler/utils.py
+++ b/src/certwrangler/utils.py
@@ -8,7 +8,6 @@
 from enum import IntEnum
 from pathlib import Path
 from string import Template
-from typing import List, Optional
 
 import yaml
 from dns.resolver import Resolver
@@ -57,7 +56,7 @@ class CertwranglerState:
     The instance of :class:`certwrangler.daemon.Daemon`.
     """
 
-    config: Optional[Config] = None
+    config: Config | None = None
     """
     The loaded instance of :class:`certwrangler.models.Config`, populated by :meth:`load_config`.
     """
@@ -73,7 +72,7 @@ class CertwranglerState:
     """
 
     _log_level: LogLevels = field(init=False, repr=False, default=LogLevels.info)
-    _loggers: List[logging.Logger] = field(
+    _loggers: list[logging.Logger] = field(
         init=False,
         repr=False,
         default_factory=lambda: [