MAINT: Enable ruff UP007/UP045 and modernise Optional/Union to PEP 604

doc/getting_started/troubleshooting/deploy_hf_model_aml.ipynb

@@ -128,14 +128,12 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "from typing import Union\n",
-    "\n",
     "from azure.ai.ml import MLClient\n",
     "from azure.core.exceptions import ResourceNotFoundError\n",
     "from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential\n",
     "\n",
     "try:\n",
-    "    credential: Union[DefaultAzureCredential, InteractiveBrowserCredential] = DefaultAzureCredential()\n",
+    "    credential: DefaultAzureCredential | InteractiveBrowserCredential = DefaultAzureCredential()\n",
     "    credential.get_token(\"https://management.azure.com/.default\")\n",
     "except Exception as ex:\n",
     "    credential = InteractiveBrowserCredential()\n",

doc/getting_started/troubleshooting/deploy_hf_model_aml.py

@@ -106,14 +106,13 @@
 # Set up the `DefaultAzureCredential` for seamless authentication with Azure services. This method should handle most authentication scenarios. If you encounter issues, refer to the [Azure Identity documentation](https://docs.microsoft.com/en-us/python/api/azure-identity/azure.identity?view=azure-python) for alternative credentials.
 #
 # %%
-from typing import Union
 
 from azure.ai.ml import MLClient
 from azure.core.exceptions import ResourceNotFoundError
 from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential
 
 try:
-    credential: Union[DefaultAzureCredential, InteractiveBrowserCredential] = DefaultAzureCredential()
+    credential: DefaultAzureCredential | InteractiveBrowserCredential = DefaultAzureCredential()
     credential.get_token("https://management.azure.com/.default")
 except Exception as ex:
     credential = InteractiveBrowserCredential()

doc/getting_started/troubleshooting/download_and_register_hf_model_aml.ipynb

@@ -71,7 +71,6 @@
    "source": [
     "# Import the Azure ML SDK components required for workspace connection and model management.\n",
     "import os\n",
-    "from typing import Union\n",
     "\n",
     "# Import necessary libraries for Azure ML operations and authentication\n",
     "from azure.ai.ml import MLClient, UserIdentityConfiguration\n",
@@ -201,7 +200,7 @@
    "source": [
     "# Setup Azure credentials, preferring DefaultAzureCredential and falling back to InteractiveBrowserCredential if necessary\n",
     "try:\n",
-    "    credential: Union[DefaultAzureCredential, InteractiveBrowserCredential] = DefaultAzureCredential()\n",
+    "    credential: DefaultAzureCredential | InteractiveBrowserCredential = DefaultAzureCredential()\n",
     "    # Verify if the default credential can fetch a token successfully\n",
     "    credential.get_token(\"https://management.azure.com/.default\")\n",
     "except Exception as ex:\n",

doc/getting_started/troubleshooting/download_and_register_hf_model_aml.py

@@ -61,7 +61,6 @@
 # %%
 # Import the Azure ML SDK components required for workspace connection and model management.
 import os
-from typing import Union
 
 # Import necessary libraries for Azure ML operations and authentication
 from azure.ai.ml import MLClient, UserIdentityConfiguration
@@ -160,7 +159,7 @@
 # %%
 # Setup Azure credentials, preferring DefaultAzureCredential and falling back to InteractiveBrowserCredential if necessary
 try:
-    credential: Union[DefaultAzureCredential, InteractiveBrowserCredential] = DefaultAzureCredential()
+    credential: DefaultAzureCredential | InteractiveBrowserCredential = DefaultAzureCredential()
     # Verify if the default credential can fetch a token successfully
     credential.get_token("https://management.azure.com/.default")
 except Exception as ex:

pyproject.toml

@@ -335,8 +335,6 @@ ignore = [
     "DOC502",  # Raised exception is not explicitly raised
     "PERF203",  # try-except-in-loop (intentional per-item error handling)
     "SIM117",  # multiple-with-statements (combining often exceeds line length)
-    "UP007",  # non-pep604-annotation-union (keep Union[X, Y] syntax)
-    "UP045",  # non-pep604-annotation-optional (keep Optional[X] syntax)
 ]
 extend-select = [
     "D204",  # 1 blank line required after class docstring

pyrit/analytics/result_analysis.py

@@ -4,7 +4,7 @@
 from collections import defaultdict
 from collections.abc import Sequence
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from pyrit.models import (
     AttackOutcome,
@@ -22,7 +22,7 @@
 class AttackStats:
     """Statistics for attack analysis results."""
 
-    success_rate: Optional[float]
+    success_rate: float | None
     total_decided: int
     successes: int
     failures: int
@@ -118,7 +118,7 @@ def get_cached_results_for_technique(
     *,
     technique_eval_hash: str,
     objective_target_eval_hash: str,
-    additional_filters: Optional[Sequence[IdentifierFilter]] = None,
+    additional_filters: Sequence[IdentifierFilter] | None = None,
 ) -> list[AttackResult]:
     """
     Return cached AttackResults matching a (technique × objective target) pair.
@@ -170,7 +170,7 @@ def get_cached_results_for_technique(
     return matches
 
 
-def _objective_target_eval_hash_for(attack_result: AttackResult) -> Optional[str]:
+def _objective_target_eval_hash_for(attack_result: AttackResult) -> str | None:
     """
     Return the ObjectiveTargetEvaluationIdentifier eval hash for a result.
 

pyrit/auth/azure_auth.py

@@ -6,7 +6,7 @@
 import inspect
 import logging
 import time
-from typing import TYPE_CHECKING, Any, Union, cast
+from typing import TYPE_CHECKING, Any, cast
 
 import msal
 from azure.core.credentials import AccessToken
@@ -41,7 +41,7 @@ class TokenProviderCredential:
     get_azure_token_provider) and Azure SDK clients that require a TokenCredential object.
     """
 
-    def __init__(self, token_provider: Callable[[], Union[str, Callable[..., Any]]]) -> None:
+    def __init__(self, token_provider: Callable[[], str | Callable[..., Any]]) -> None:
         """
         Initialize TokenProviderCredential.
 
@@ -75,7 +75,7 @@ class AsyncTokenProviderCredential:
     async clients that require an AsyncTokenCredential object (with async def get_token).
     """
 
-    def __init__(self, token_provider: Callable[[], Union[str, Awaitable[str]]]) -> None:
+    def __init__(self, token_provider: Callable[[], str | Awaitable[str]]) -> None:
         """
         Initialize AsyncTokenProviderCredential.
 
@@ -394,7 +394,7 @@ def get_azure_openai_auth(endpoint: str) -> Callable[[], Awaitable[str]]:
     return get_azure_async_token_provider(scope)
 
 
-def get_speech_config(resource_id: Union[str, None], key: Union[str, None], region: str) -> speechsdk.SpeechConfig:
+def get_speech_config(resource_id: str | None, key: str | None, region: str) -> speechsdk.SpeechConfig:
     """
     Get the speech config using key/region pair (for key auth scenarios) or resource_id/region pair
     (for Entra auth scenarios).
@@ -436,8 +436,8 @@ def get_speech_config(resource_id: Union[str, None], key: Union[str, None], regi
 async def get_speech_config_async(
     *,
     token_provider: Callable[[], str | Awaitable[str]] | None,
-    resource_id: Union[str, None],
-    key: Union[str, None],
+    resource_id: str | None,
+    key: str | None,
     region: str,
 ) -> speechsdk.SpeechConfig:
     """

pyrit/auth/copilot_authenticator.py

@@ -7,7 +7,7 @@
 import os
 import sys
 from datetime import datetime, timedelta, timezone
-from typing import Any, Optional
+from typing import Any
 
 from msal_extensions import FilePersistence, build_encrypted_persistence
 
@@ -196,7 +196,7 @@ def _create_persistent_cache(cache_file: str, fallback_to_plaintext: bool = Fals
             logger.error(f"Encryption unavailable ({e}) and fallback_to_plaintext is False. Cannot proceed.")
             raise
 
-    async def _get_cached_token_if_available_and_valid_async(self) -> Optional[dict[str, Any]]:
+    async def _get_cached_token_if_available_and_valid_async(self) -> dict[str, Any] | None:
         """
         Retrieve and validate cached token.
 
@@ -258,7 +258,7 @@ async def _get_cached_token_if_available_and_valid_async(self) -> Optional[dict[
                 logger.error(f"Failed to load cached token ({error_name}): {e}")
             return None
 
-    def _save_token_to_cache(self, *, token: str, expires_in: Optional[int] = None) -> None:
+    def _save_token_to_cache(self, *, token: str, expires_in: int | None = None) -> None:
         """
         Save token to persistent cache with metadata.
 
@@ -301,7 +301,7 @@ def _clear_token_cache(self) -> None:
         except Exception as e:
             logger.error(f"Failed to clear cache: {e}")
 
-    async def _fetch_access_token_with_playwright_async(self) -> Optional[str]:
+    async def _fetch_access_token_with_playwright_async(self) -> str | None:
         """
         Fetch access token using Playwright browser automation.
 
@@ -339,7 +339,7 @@ async def _fetch_access_token_with_playwright_async(self) -> Optional[str]:
         # If not on Windows or using the right loop already, proceed normally
         return await self._run_playwright_browser_automation_async()
 
-    async def _run_playwright_in_thread_async(self) -> Optional[str]:
+    async def _run_playwright_in_thread_async(self) -> str | None:
         """
         Run Playwright browser automation in a separate thread with ProactorEventLoop.
         This is needed on Windows when the main loop is SelectorEventLoop (e.g., in Jupyter).
@@ -348,21 +348,21 @@ async def _run_playwright_in_thread_async(self) -> Optional[str]:
             Optional[str]: The bearer token if successfully retrieved, None otherwise.
         """
 
-        def run_in_new_loop() -> Optional[str]:
+        def run_in_new_loop() -> str | None:
             if sys.platform == "win32":
                 new_loop = asyncio.ProactorEventLoop()
             else:
                 new_loop = asyncio.new_event_loop()
             asyncio.set_event_loop(new_loop)
             try:
-                result: Optional[str] = new_loop.run_until_complete(self._run_playwright_browser_automation_async())
+                result: str | None = new_loop.run_until_complete(self._run_playwright_browser_automation_async())
                 return result
             finally:
                 new_loop.close()
 
         return await asyncio.get_running_loop().run_in_executor(None, run_in_new_loop)
 
-    async def _run_playwright_browser_automation_async(self) -> Optional[str]:
+    async def _run_playwright_browser_automation_async(self) -> str | None:
         """
         Execute the actual Playwright browser automation to fetch the access token.
 

pyrit/auth/manual_copilot_authenticator.py

@@ -3,7 +3,7 @@
 
 import logging
 import os
-from typing import Any, Optional
+from typing import Any
 
 import jwt
 
@@ -36,7 +36,7 @@ class ManualCopilotAuthenticator(Authenticator):
     #: Environment variable name for the Copilot access token
     ACCESS_TOKEN_ENV_VAR: str = "COPILOT_ACCESS_TOKEN"
 
-    def __init__(self, *, access_token: Optional[str] = None) -> None:
+    def __init__(self, *, access_token: str | None = None) -> None:
         """
         Initialize the ManualCopilotAuthenticator with a pre-obtained access token.