MAINT Enable ruff UP (pyupgrade) rules and fix all violations

build_scripts/check_links.py

@@ -34,7 +34,7 @@
 
 
 def extract_urls(file_path):
-    with open(file_path, "r", encoding="utf-8") as file:
+    with open(file_path, encoding="utf-8") as file:
         content = file.read()
     matches = URL_PATTERN.findall(content)
     # Flatten the list of tuples and filter out empty strings

build_scripts/generate_rss.py

@@ -60,7 +60,7 @@ def handle_data(self, data):
     fe.guid(f"https://azure.github.io/PyRIT/blog/{file.name}")
 
     # Extract title and description from HTML content
-    with open(file, "r", encoding="utf-8") as f:
+    with open(file, encoding="utf-8") as f:
         parser = BlogEntryParser()
         parser.feed(f.read())
         fe.title(parser.title)

build_scripts/remove_notebook_headers.py

@@ -10,7 +10,7 @@ def remove_kernelspec_from_ipynb_files(file_path: str):
 
     if file_path.endswith(".ipynb"):
         # Iterate through all .ipynb files in the specified file
-        with open(file_path, "r", encoding="utf-8") as f:
+        with open(file_path, encoding="utf-8") as f:
             content = json.load(f)
             # Remove the "kernelspec" metadata section if it exists
             if "metadata" in content and "kernelspec" in content["metadata"]:

build_scripts/validate_jupyter_book.py

@@ -20,19 +20,18 @@
 import re
 import sys
 from pathlib import Path
-from typing import List, Set, Tuple
 
 import yaml
 
 
-def parse_api_rst(api_rst_path: Path) -> List[Tuple[str, List[str]]]:
+def parse_api_rst(api_rst_path: Path) -> list[tuple[str, list[str]]]:
     """
     Parse api.rst file to extract module names and their autosummary members.
 
     Returns:
         List of tuples: (module_name, [member_names])
     """
-    with open(api_rst_path, "r", encoding="utf-8") as f:
+    with open(api_rst_path, encoding="utf-8") as f:
         content = f.read()
 
     modules = []
@@ -61,7 +60,7 @@ def parse_api_rst(api_rst_path: Path) -> List[Tuple[str, List[str]]]:
     return modules
 
 
-def validate_api_rst_modules(modules: List[Tuple[str, List[str]]], repo_root: Path) -> List[str]:
+def validate_api_rst_modules(modules: list[tuple[str, list[str]]], repo_root: Path) -> list[str]:
     """
     Validate that modules exist and autosummary members are defined.
 
@@ -110,7 +109,7 @@ def validate_api_rst_modules(modules: List[Tuple[str, List[str]]], repo_root: Pa
             if module_file:
                 # Read the source file and check for member definitions
                 try:
-                    with open(module_file, "r", encoding="utf-8") as f:
+                    with open(module_file, encoding="utf-8") as f:
                         source_content = f.read()
 
                     for member in members:
@@ -141,14 +140,14 @@ def validate_api_rst_modules(modules: List[Tuple[str, List[str]]], repo_root: Pa
     return errors
 
 
-def parse_toc_yml(toc_path: Path) -> Set[str]:
+def parse_toc_yml(toc_path: Path) -> set[str]:
     """
     Parse _toc.yml file to extract all file references.
 
     Returns:
         Set of file paths (relative to doc/ directory, without extensions)
     """
-    with open(toc_path, "r", encoding="utf-8") as f:
+    with open(toc_path, encoding="utf-8") as f:
         toc_data = yaml.safe_load(f)
 
     files = set()
@@ -169,7 +168,7 @@ def extract_files(node):
     return files
 
 
-def validate_toc_yml_files(toc_files: Set[str], doc_root: Path) -> List[str]:
+def validate_toc_yml_files(toc_files: set[str], doc_root: Path) -> list[str]:
     """
     Validate that all files referenced in _toc.yml exist.
 
@@ -198,7 +197,7 @@ def validate_toc_yml_files(toc_files: Set[str], doc_root: Path) -> List[str]:
     return errors
 
 
-def find_orphaned_doc_files(toc_files: Set[str], doc_root: Path) -> List[str]:
+def find_orphaned_doc_files(toc_files: set[str], doc_root: Path) -> list[str]:
     """
     Find documentation files that exist but are not referenced in _toc.yml.
 

doc/code/executor/workflow/1_xpia_website.ipynb

@@ -30,7 +30,7 @@
     "from pyrit.models import Message, MessagePiece\n",
     "\n",
     "# Read basic HTML file with template slot for the XPIA.\n",
-    "with open(Path().cwd() / \"example\" / \"index.html\", \"r\") as f:\n",
+    "with open(Path().cwd() / \"example\" / \"index.html\") as f:\n",
     "    html_template = f.read()\n",
     "jailbreak_template = TextJailBreak(string_template=html_template)\n",
     "\n",

doc/code/executor/workflow/1_xpia_website.py

@@ -25,7 +25,7 @@
 from pyrit.models import Message, MessagePiece
 
 # Read basic HTML file with template slot for the XPIA.
-with open(Path().cwd() / "example" / "index.html", "r") as f:
+with open(Path().cwd() / "example" / "index.html") as f:
     html_template = f.read()
 jailbreak_template = TextJailBreak(string_template=html_template)
 

doc/code/scenarios/0_scenarios.ipynb

@@ -96,7 +96,7 @@
     }
    ],
    "source": [
-    "from typing import List, Optional, Type\n",
+    "from typing import Optional\n",
     "\n",
     "from pyrit.common import apply_defaults\n",
     "from pyrit.executor.attack import AttackScoringConfig, PromptSendingAttack\n",
@@ -124,7 +124,7 @@
     "\n",
     "    # A strategy defintion helps callers define how to run your scenario (e.g. from the front_end)\n",
     "    @classmethod\n",
-    "    def get_strategy_class(cls) -> Type[ScenarioStrategy]:\n",
+    "    def get_strategy_class(cls) -> type[ScenarioStrategy]:\n",
     "        return MyStrategy\n",
     "\n",
     "    @classmethod\n",
@@ -155,7 +155,7 @@
     "            scenario_result_id=scenario_result_id,\n",
     "        )\n",
     "\n",
-    "    async def _get_atomic_attacks_async(self) -> List[AtomicAttack]:\n",
+    "    async def _get_atomic_attacks_async(self) -> list[AtomicAttack]:\n",
     "        \"\"\"\n",
     "        Build atomic attacks based on selected strategies.\n",
     "\n",

doc/code/scenarios/0_scenarios.py

@@ -84,7 +84,7 @@
 #
 # ### Example Structure
 # %%
-from typing import List, Optional, Type
+from typing import Optional
 
 from pyrit.common import apply_defaults
 from pyrit.executor.attack import AttackScoringConfig, PromptSendingAttack
@@ -112,7 +112,7 @@ class MyScenario(Scenario):
 
     # A strategy defintion helps callers define how to run your scenario (e.g. from the front_end)
     @classmethod
-    def get_strategy_class(cls) -> Type[ScenarioStrategy]:
+    def get_strategy_class(cls) -> type[ScenarioStrategy]:
         return MyStrategy
 
     @classmethod
@@ -143,7 +143,7 @@ def __init__(
             scenario_result_id=scenario_result_id,
         )
 
-    async def _get_atomic_attacks_async(self) -> List[AtomicAttack]:
+    async def _get_atomic_attacks_async(self) -> list[AtomicAttack]:
         """
         Build atomic attacks based on selected strategies.
 

doc/code/targets/11_message_normalizer.ipynb

@@ -491,16 +491,14 @@
     }
    ],
    "source": [
-    "from typing import List\n",
-    "\n",
     "from pyrit.message_normalizer import MessageStringNormalizer\n",
     "from pyrit.models import Message\n",
     "\n",
     "\n",
     "class SimpleMarkdownNormalizer(MessageStringNormalizer):\n",
     "    \"\"\"Custom normalizer that formats messages as Markdown.\"\"\"\n",
     "\n",
-    "    async def normalize_string_async(self, messages: List[Message]) -> str:\n",
+    "    async def normalize_string_async(self, messages: list[Message]) -> str:\n",
     "        lines = []\n",
     "        for msg in messages:\n",
     "            piece = msg.get_piece()\n",

doc/code/targets/11_message_normalizer.py

@@ -203,7 +203,6 @@
 # You can create custom normalizers by extending the base classes.
 
 # %%
-from typing import List
 
 from pyrit.message_normalizer import MessageStringNormalizer
 from pyrit.models import Message
@@ -212,7 +211,7 @@
 class SimpleMarkdownNormalizer(MessageStringNormalizer):
     """Custom normalizer that formats messages as Markdown."""
 
-    async def normalize_string_async(self, messages: List[Message]) -> str:
+    async def normalize_string_async(self, messages: list[Message]) -> str:
         lines = []
         for msg in messages:
             piece = msg.get_piece()

doc/code/targets/6_custom_targets.ipynb

@@ -254,8 +254,6 @@
     }
    ],
    "source": [
-    "from typing import List\n",
-    "\n",
     "from pyrit.executor.attack import (\n",
     "    AttackConverterConfig,\n",
     "    AttackScoringConfig,\n",
@@ -281,7 +279,7 @@
     "\n",
     "aoai_target = OpenAIChatTarget()\n",
     "\n",
-    "converters: List[PromptConverterConfiguration] = PromptConverterConfiguration.from_converters(\n",
+    "converters: list[PromptConverterConfiguration] = PromptConverterConfiguration.from_converters(\n",
     "    converters=[RandomCapitalLettersConverter(percentage=25)]\n",
     ")\n",
     "\n",

doc/code/targets/6_custom_targets.py

@@ -103,7 +103,6 @@
 # Below is an example of using PromptSendingAttack, which allows the use of all our converters. For example, you could use this to utilize all the built-in jailbreaks, base64 encode them, use variations, different languages, etc.
 
 # %%
-from typing import List
 
 from pyrit.executor.attack import (
     AttackConverterConfig,
@@ -130,7 +129,7 @@
 
 aoai_target = OpenAIChatTarget()
 
-converters: List[PromptConverterConfiguration] = PromptConverterConfiguration.from_converters(
+converters: list[PromptConverterConfiguration] = PromptConverterConfiguration.from_converters(
     converters=[RandomCapitalLettersConverter(percentage=25)]
 )
 

doc/cookbooks/5_psychosocial_harms.ipynb

@@ -195,7 +195,7 @@
     "print(\"Attack Technique using Escalation for a user in imminent crisis:\")\n",
     "attack_strategy_path = pathlib.Path(DATASETS_PATH) / \"executors\" / \"crescendo\" / \"escalation_crisis.yaml\"\n",
     "\n",
-    "with open(attack_strategy_path, \"r\") as file:\n",
+    "with open(attack_strategy_path) as file:\n",
     "    print(file.read())"
    ]
   },

doc/cookbooks/5_psychosocial_harms.py

@@ -77,7 +77,7 @@
 print("Attack Technique using Escalation for a user in imminent crisis:")
 attack_strategy_path = pathlib.Path(DATASETS_PATH) / "executors" / "crescendo" / "escalation_crisis.yaml"
 
-with open(attack_strategy_path, "r") as file:
+with open(attack_strategy_path) as file:
     print(file.read())
 
 # %% [markdown]

doc/deployment/deploy_hf_model_aml.ipynb

@@ -186,16 +186,12 @@
     "if check_model_version_exists(workspace_ml_client, model_to_deploy, model_version):\n",
     "    print(\"Model found in the Azure ML workspace model registry.\")\n",
     "    model = workspace_ml_client.models.get(model_to_deploy, model_version)\n",
-    "    print(\n",
-    "        \"\\n\\nUsing model name: {0}, version: {1}, id: {2} for inferencing\".format(model.name, model.version, model.id)\n",
-    "    )\n",
+    "    print(f\"\\n\\nUsing model name: {model.name}, version: {model.version}, id: {model.id} for inferencing\")\n",
     "# Check if the Hugging Face model exists in the Azure ML model catalog registry\n",
     "elif check_model_version_exists(registry_ml_client, model_to_deploy, model_version):\n",
     "    print(\"Model found in the Azure ML model catalog registry.\")\n",
     "    model = registry_ml_client.models.get(model_to_deploy, model_version)\n",
-    "    print(\n",
-    "        \"\\n\\nUsing model name: {0}, version: {1}, id: {2} for inferencing\".format(model.name, model.version, model.id)\n",
-    "    )\n",
+    "    print(f\"\\n\\nUsing model name: {model.name}, version: {model.version}, id: {model.id} for inferencing\")\n",
     "else:\n",
     "    raise ValueError(\n",
     "        f\"Model {model_to_deploy} not found in any registry. Please run the notebook (download_and_register_hf_model_aml.ipynb) to download and register Hugging Face model to Azure ML workspace model registry.\"\n",

doc/deployment/deploy_hf_model_aml.py

@@ -154,16 +154,12 @@ def check_model_version_exists(client, model_name, version) -> bool:
 if check_model_version_exists(workspace_ml_client, model_to_deploy, model_version):
     print("Model found in the Azure ML workspace model registry.")
     model = workspace_ml_client.models.get(model_to_deploy, model_version)
-    print(
-        "\n\nUsing model name: {0}, version: {1}, id: {2} for inferencing".format(model.name, model.version, model.id)
-    )
+    print(f"\n\nUsing model name: {model.name}, version: {model.version}, id: {model.id} for inferencing")
 # Check if the Hugging Face model exists in the Azure ML model catalog registry
 elif check_model_version_exists(registry_ml_client, model_to_deploy, model_version):
     print("Model found in the Azure ML model catalog registry.")
     model = registry_ml_client.models.get(model_to_deploy, model_version)
-    print(
-        "\n\nUsing model name: {0}, version: {1}, id: {2} for inferencing".format(model.name, model.version, model.id)
-    )
+    print(f"\n\nUsing model name: {model.name}, version: {model.version}, id: {model.id} for inferencing")
 else:
     raise ValueError(
         f"Model {model_to_deploy} not found in any registry. Please run the notebook (download_and_register_hf_model_aml.ipynb) to download and register Hugging Face model to Azure ML workspace model registry."

doc/generate_docs/pct_to_ipynb.py

@@ -52,7 +52,7 @@ def main():
     cache_file = os.path.join(cache_dir, f"pct_to_ipynb_{args.run_id}.cache")
     processed_files = set()
     if os.path.isfile(cache_file):
-        with open(cache_file, "r") as f:
+        with open(cache_file) as f:
             for file_path in f:
                 processed_files.add(file_path.strip())
 

frontend/dev.py

@@ -42,7 +42,7 @@ def sync_version():
 
         # Read package.json
         package_json_path = FRONTEND_DIR / "package.json"
-        with open(package_json_path, "r") as f:
+        with open(package_json_path) as f:
             package_data = json.load(f)
 
         # Update version if different

pyproject.toml

@@ -253,6 +253,7 @@ select = [
     "F401",  # unused-import
     "I",  # isort
     "RET",  # https://docs.astral.sh/ruff/rules/#flake8-return-ret
+    "UP",  # https://docs.astral.sh/ruff/rules/#pyupgrade-up
     "W",  # https://docs.astral.sh/ruff/rules/#pycodestyle-w
 ]
 ignore = [
@@ -262,6 +263,8 @@ ignore = [
     "D212",  # Multi-line docstring summary should start at the first line
     "D301",  # Use r""" if any backslashes in a docstring
     "DOC502",  # Raised exception is not explicitly raised
+    "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

@@ -3,7 +3,7 @@
 
 from collections import defaultdict
 from dataclasses import dataclass
-from typing import DefaultDict, Optional
+from typing import Optional
 
 from pyrit.models import AttackOutcome, AttackResult
 
@@ -54,8 +54,8 @@ def analyze_results(attack_results: list[AttackResult]) -> dict[str, AttackStats
     if not attack_results:
         raise ValueError("attack_results cannot be empty")
 
-    overall_counts: DefaultDict[str, int] = defaultdict(int)
-    by_type_counts: DefaultDict[str, DefaultDict[str, int]] = defaultdict(lambda: defaultdict(int))
+    overall_counts: defaultdict[str, int] = defaultdict(int)
+    by_type_counts: defaultdict[str, defaultdict[str, int]] = defaultdict(lambda: defaultdict(int))
 
     for attack in attack_results:
         if not isinstance(attack, AttackResult):

pyrit/auth/azure_auth.py

@@ -5,7 +5,8 @@
 
 import logging
 import time
-from typing import TYPE_CHECKING, Any, Callable, Union, cast
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any, Union, cast
 from urllib.parse import urlparse
 
 import msal