MAINT Enable ruff PERF rule for performance linting

pyproject.toml

@@ -234,6 +234,7 @@ fixable = [
     "ISC",
     "NPY",
     "PD",
+    "PERF",
     "PGH",
     "PIE",
     "PL",
@@ -259,6 +260,7 @@ select = [
     "DOC",  # https://docs.astral.sh/ruff/rules/#pydoclint-doc
     "F401",  # unused-import
     "I",  # isort
+    "PERF",  # https://docs.astral.sh/ruff/rules/#perflint-perf
     "PIE",  # https://docs.astral.sh/ruff/rules/#flake8-pie-pie
     "RET",  # https://docs.astral.sh/ruff/rules/#flake8-return-ret
     "SIM",  # https://docs.astral.sh/ruff/rules/#flake8-simplify-sim
@@ -274,6 +276,7 @@ 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
+    "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)

pyrit/analytics/conversation_analytics.py

@@ -39,20 +39,16 @@ def get_prompt_entries_with_same_converted_content(
             the similar chat messages based on content.
         """
         all_memories = self.memory_interface.get_message_pieces()
-        similar_messages = []
-
-        for memory in all_memories:
-            if memory.converted_value == chat_message_content:
-                similar_messages.append(
-                    ConversationMessageWithSimilarity(
-                        score=1.0,
-                        role=memory.role,
-                        content=memory.converted_value,
-                        metric="exact_match",  # Exact match
-                    )
-                )
-
-        return similar_messages
+        return [
+            ConversationMessageWithSimilarity(
+                score=1.0,
+                role=memory.role,
+                content=memory.converted_value,
+                metric="exact_match",  # Exact match
+            )
+            for memory in all_memories
+            if memory.converted_value == chat_message_content
+        ]
 
     def get_similar_chat_messages_by_embedding(
         self, *, chat_message_embedding: list[float], threshold: float = 0.8

pyrit/auxiliary_attacks/gcg/attack/base/attack_manager.py

@@ -107,10 +107,7 @@ def get_nonascii_toks(tokenizer: Any, device: str = "cpu") -> torch.Tensor:
     def is_ascii(s: str) -> bool:
         return s.isascii() and s.isprintable()
 
-    ascii_toks = []
-    for i in range(3, tokenizer.vocab_size):
-        if not is_ascii(tokenizer.decode([i])):
-            ascii_toks.append(i)
+    ascii_toks = [i for i in range(3, tokenizer.vocab_size) if not is_ascii(tokenizer.decode([i]))]
 
     if tokenizer.bos_token_id is not None:
         ascii_toks.append(tokenizer.bos_token_id)

pyrit/executor/attack/printer/markdown_printer.py

@@ -89,8 +89,7 @@ def _format_score(self, score: Score, indent: str = "") -> str:
             rationale_lines = score.score_rationale.split("\n")
             if len(rationale_lines) > 1:
                 lines.append(f"{indent}- **Rationale:**")
-                for line in rationale_lines:
-                    lines.append(f"{indent}  {line}")
+                lines.extend(f"{indent}  {line}" for line in rationale_lines)
             else:
                 lines.append(f"{indent}- **Rationale:** {score.score_rationale}")
 
@@ -273,8 +272,7 @@ def _format_system_message(self, message: Message) -> list[str]:
             List[str]: List of markdown strings representing the system message.
         """
         lines = ["\n### System Message\n"]
-        for piece in message.message_pieces:
-            lines.append(f"{piece.converted_value}\n")
+        lines.extend(f"{piece.converted_value}\n" for piece in message.message_pieces)
         return lines
 
     async def _format_user_message_async(self, *, message: Message, turn_number: int) -> list[str]:
@@ -461,8 +459,7 @@ def _format_message_scores(self, message: Message) -> list[str]:
             scores = self._memory.get_prompt_scores(prompt_ids=[str(piece.id)])
             if scores:
                 lines.append("\n##### Scores\n")
-                for score in scores:
-                    lines.append(self._format_score(score, indent=""))
+                lines.extend(self._format_score(score, indent="") for score in scores)
                 lines.append("")
         return lines
 
@@ -572,8 +569,7 @@ async def _get_pruned_conversations_markdown_async(self, result: AttackResult) -
                 scores = self._memory.get_prompt_scores(prompt_ids=[str(piece.id)])
                 if scores:
                     markdown_lines.append("\n**Score:**\n")
-                    for score in scores:
-                        markdown_lines.append(self._format_score(score, indent=""))
+                    markdown_lines.extend(self._format_score(score, indent="") for score in scores)
 
         return markdown_lines
 

pyrit/executor/promptgen/fuzzer/fuzzer.py

@@ -957,14 +957,13 @@ def _get_other_templates(self, context: FuzzerContext) -> list[str]:
         Returns:
             List of template strings.
         """
-        other_templates = []
         node_ids_on_path = {node.id for node in context.mcts_selected_path}
 
-        for prompt_node in context.initial_prompt_nodes + context.new_prompt_nodes:
-            if prompt_node.id not in node_ids_on_path:
-                other_templates.append(prompt_node.template)
-
-        return other_templates
+        return [
+            prompt_node.template
+            for prompt_node in context.initial_prompt_nodes + context.new_prompt_nodes
+            if prompt_node.id not in node_ids_on_path
+        ]
 
     def _generate_prompts_from_template(self, *, template: SeedPrompt, prompts: list[str]) -> list[str]:
         """

pyrit/identifiers/component_identifier.py

@@ -80,9 +80,7 @@ def _build_hash_dict(
 
     # Only include non-None params — adding an optional param with None default
     # won't change existing hashes, making the schema backward-compatible.
-    for key, value in sorted(params.items()):
-        if value is not None:
-            hash_dict[key] = value
+    hash_dict.update({key: value for key, value in sorted(params.items()) if value is not None})
 
     # Children contribute their hashes, not their full structure.
     if children:

pyrit/memory/memory_exporter.py

@@ -69,9 +69,7 @@ def export_to_json(self, data: list[MessagePiece], file_path: Optional[Path] = N
             raise ValueError("Please provide a valid file path for exporting data.")
         if not data:
             raise ValueError("No data to export.")
-        export_data = []
-        for piece in data:
-            export_data.append(piece.to_dict())
+        export_data = [piece.to_dict() for piece in data]
         with open(file_path, "w") as f:
             json.dump(export_data, f, indent=4)
 
@@ -92,9 +90,7 @@ def export_to_csv(self, data: list[MessagePiece], file_path: Optional[Path] = No
             raise ValueError("Please provide a valid file path for exporting data.")
         if not data:
             raise ValueError("No data to export.")
-        export_data = []
-        for piece in data:
-            export_data.append(piece.to_dict())
+        export_data = [piece.to_dict() for piece in data]
         fieldnames = list(export_data[0].keys())
         with open(file_path, "w", newline="") as f:
             writer = csv.DictWriter(f, fieldnames=fieldnames)

pyrit/memory/memory_interface.py

@@ -954,8 +954,7 @@ def _add_list_conditions(
         self, field: InstrumentedAttribute[Any], conditions: list[Any], values: Optional[Sequence[str]] = None
     ) -> None:
         if values:
-            for value in values:
-                conditions.append(field.contains(value))
+            conditions.extend(field.contains(value) for value in values)
 
     async def _serialize_seed_value(self, prompt: Seed) -> str:
         """

pyrit/models/scenario_result.py

@@ -151,8 +151,7 @@ def get_objectives(self, *, atomic_attack_name: Optional[str] = None) -> list[st
                 strategies_to_process = []
 
         for results in strategies_to_process:
-            for result in results:
-                objectives.append(result.objective)
+            objectives.extend(result.objective for result in results)
 
         return list(set(objectives))
 

pyrit/prompt_converter/audio_echo_converter.py

@@ -117,9 +117,7 @@ async def convert_async(self, *, prompt: str, input_type: PromptDataType = "audi
             if data.ndim == 1:
                 echo_data = self._apply_echo(data, sample_rate).astype(original_dtype)
             else:
-                channels = []
-                for ch in range(data.shape[1]):
-                    channels.append(self._apply_echo(data[:, ch], sample_rate))
+                channels = [self._apply_echo(data[:, ch], sample_rate) for ch in range(data.shape[1])]
                 echo_data = np.column_stack(channels).astype(original_dtype)
 
             # Write the processed data as a new WAV file

pyrit/prompt_converter/audio_volume_converter.py

@@ -112,9 +112,7 @@ async def convert_async(self, *, prompt: str, input_type: PromptDataType = "audi
                 volume_data = self._apply_volume(data).astype(original_dtype)
             else:
                 # Multi-channel audio (e.g., stereo)
-                channels = []
-                for ch in range(data.shape[1]):
-                    channels.append(self._apply_volume(data[:, ch]))
+                channels = [self._apply_volume(data[:, ch]) for ch in range(data.shape[1])]
                 volume_data = np.column_stack(channels).astype(original_dtype)
 
             # Write the processed data as a new WAV file

pyrit/prompt_converter/audio_white_noise_converter.py

@@ -114,9 +114,7 @@ async def convert_async(self, *, prompt: str, input_type: PromptDataType = "audi
             if data.ndim == 1:
                 noisy_data = self._add_noise(data).astype(original_dtype)
             else:
-                channels = []
-                for ch in range(data.shape[1]):
-                    channels.append(self._add_noise(data[:, ch]))
+                channels = [self._add_noise(data[:, ch]) for ch in range(data.shape[1])]
                 noisy_data = np.column_stack(channels).astype(original_dtype)
 
             # Write the processed data as a new WAV file

pyrit/prompt_converter/nato_converter.py

@@ -90,9 +90,6 @@ def _convert_to_nato(self, text: str) -> str:
         Returns:
             str: The NATO phonetic alphabet representation, with code words separated by spaces.
         """
-        output = []
-        for char in text.upper():
-            if char in self._NATO_MAP:
-                output.append(self._NATO_MAP[char])
+        output = [self._NATO_MAP[char] for char in text.upper() if char in self._NATO_MAP]
 
         return " ".join(output)

pyrit/prompt_target/openai/openai_chat_target.py

@@ -644,8 +644,7 @@ async def _construct_request_body(
         }
 
         if self._extra_body_parameters:
-            for key, value in self._extra_body_parameters.items():
-                body_parameters[key] = value
+            body_parameters.update(self._extra_body_parameters)
 
         # Filter out None values
         return {k: v for k, v in body_parameters.items() if v is not None}

pyrit/prompt_target/openai/openai_response_target.py

@@ -256,9 +256,7 @@ async def _build_input_for_multi_modal_async(self, conversation: MutableSequence
 
             # System message (remapped to developer)
             if pieces[0].api_role == "system":
-                system_content = []
-                for piece in pieces:
-                    system_content.append({"type": "input_text", "text": piece.converted_value})
+                system_content = [{"type": "input_text", "text": piece.converted_value} for piece in pieces]
                 input_items.append({"role": "developer", "content": system_content})
                 continue
 

pyrit/scenario/core/scenario_strategy.py

@@ -556,8 +556,9 @@ def normalize_compositions(
                 aggregate = aggregates_in_composition[0]
                 expanded = strategy_type.normalize_strategies({aggregate})
                 # Each expanded strategy becomes its own composition
-                for strategy in expanded:
-                    normalized_compositions.append(ScenarioCompositeStrategy(strategies=[strategy]))
+                normalized_compositions.extend(
+                    ScenarioCompositeStrategy(strategies=[strategy]) for strategy in expanded
+                )
             else:
                 # Concrete composition - validate and preserve as-is
                 strategy_type.validate_composition(typed_strategies)

pyrit/scenario/scenarios/airt/cyber.py

@@ -288,11 +288,8 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]:
         # Resolve seed groups from deprecated objectives or dataset config
         self._seed_groups = self._resolve_seed_groups()
 
-        atomic_attacks: list[AtomicAttack] = []
         strategies = ScenarioCompositeStrategy.extract_single_strategy_values(
             composites=self._scenario_composites, strategy_type=CyberStrategy
         )
 
-        for strategy in strategies:
-            atomic_attacks.append(self._get_atomic_attack_from_strategy(strategy))
-        return atomic_attacks
+        return [self._get_atomic_attack_from_strategy(strategy) for strategy in strategies]

pyrit/scenario/scenarios/airt/leakage_scenario.py

@@ -373,11 +373,8 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]:
         # Resolve objectives to seed groups format
         self._seed_groups = self._resolve_seed_groups()
 
-        atomic_attacks: list[AtomicAttack] = []
         strategies = ScenarioCompositeStrategy.extract_single_strategy_values(
             composites=self._scenario_composites, strategy_type=LeakageStrategy
         )
 
-        for strategy in strategies:
-            atomic_attacks.append(await self._get_atomic_attack_from_strategy_async(strategy))
-        return atomic_attacks
+        return [await self._get_atomic_attack_from_strategy_async(strategy) for strategy in strategies]

pyrit/scenario/scenarios/airt/scam.py

@@ -325,12 +325,8 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]:
         # Resolve seed groups from deprecated objectives or dataset config
         self._seed_groups = self._resolve_seed_groups()
 
-        atomic_attacks: list[AtomicAttack] = []
         strategies = ScenarioCompositeStrategy.extract_single_strategy_values(
             composites=self._scenario_composites, strategy_type=ScamStrategy
         )
 
-        for strategy in strategies:
-            atomic_attacks.append(self._get_atomic_attack_from_strategy(strategy))
-
-        return atomic_attacks
+        return [self._get_atomic_attack_from_strategy(strategy) for strategy in strategies]

pyrit/scenario/scenarios/foundry/red_team_agent.py

@@ -345,10 +345,7 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]:
         # Resolve seed groups now that initialize_async has been called
         self._seed_groups = self._resolve_seed_groups()
 
-        atomic_attacks = []
-        for composition in self._scenario_composites:
-            atomic_attacks.append(self._get_attack_from_strategy(composition))
-        return atomic_attacks
+        return [self._get_attack_from_strategy(composition) for composition in self._scenario_composites]
 
     def _get_default_adversarial_target(self) -> OpenAIChatTarget:
         return OpenAIChatTarget(

pyrit/scenario/scenarios/garak/encoding.py

@@ -241,11 +241,7 @@ def _resolve_seed_groups(self) -> list[SeedAttackGroup]:
 
         # Use deprecated seed_prompts if provided
         if self._deprecated_seed_prompts is not None:
-            seed_groups = []
-            for seed in self._deprecated_seed_prompts:
-                seed_groups.append(SeedAttackGroup(seeds=[SeedObjective(value=seed)]))
-
-            return seed_groups
+            return [SeedAttackGroup(seeds=[SeedObjective(value=seed)]) for seed in self._deprecated_seed_prompts]
 
         # Use dataset_config (guaranteed to be set by initialize_async)
         seed_groups = self._dataset_config.get_all_seed_attack_groups()

pyrit/score/scorer.py

@@ -783,17 +783,15 @@ async def score_response_multiple_scorers_async(
             return []
 
         # Create all scoring tasks, note TEMPORARY fix to prevent multi-piece responses from breaking scoring logic
-        tasks = []
-
-        for scorer in scorers:
-            tasks.append(
-                scorer.score_async(
-                    message=response,
-                    objective=objective,
-                    role_filter=role_filter,
-                    skip_on_error_result=skip_on_error_result,
-                )
+        tasks = [
+            scorer.score_async(
+                message=response,
+                objective=objective,
+                role_filter=role_filter,
+                skip_on_error_result=skip_on_error_result,
             )
+            for scorer in scorers
+        ]
 
         if not tasks:
             return []

pyrit/score/scorer_evaluation/scorer_metrics_io.py

@@ -67,9 +67,13 @@ def _build_eval_dict(
         ComponentIdentifier.KEY_CLASS_MODULE: identifier.class_module,
     }
 
-    for key, value in sorted(identifier.params.items()):
-        if value is not None and (param_allowlist is None or key in param_allowlist):
-            eval_dict[key] = value
+    eval_dict.update(
+        {
+            k: v
+            for k, v in sorted(identifier.params.items())
+            if v is not None and (param_allowlist is None or k in param_allowlist)
+        }
+    )
 
     if identifier.children:
         eval_children: dict[str, Any] = {}

pyrit/setup/initializers/pyrit_initializer.py

@@ -153,7 +153,7 @@ def _track_initialization_changes(self) -> Iterator[dict[str, Any]]:
             new_main_dict = sys.modules["__main__"].__dict__
 
             # Track default values that were added - just collect class.parameter pairs
-            for scope, _value in new_defaults.items():
+            for scope in new_defaults:
                 if scope not in current_default_keys:
                     class_param = f"{scope.class_type.__name__}.{scope.parameter_name}"
                     if class_param not in tracking_info["default_values"]:

tests/unit/datasets/test_jailbreak_text.py

@@ -223,7 +223,7 @@ def teardown_method(self) -> None:
     def test_scan_template_files_excludes_multi_parameter(self) -> None:
         """Test that _scan_template_files excludes files under multi_parameter directories."""
         result = TextJailBreak._scan_template_files()
-        for _filename, paths in result.items():
+        for paths in result.values():
             for path in paths:
                 assert "multi_parameter" not in path.parts
 

tests/unit/executor/attack/test_attack_parameter_consistency.py

@@ -912,9 +912,9 @@ def _get_adversarial_chat_text_values(*, adversarial_chat_conversation_id: str)
 
     text_values = []
     for msg in conversation:
-        for piece in msg.message_pieces:
-            if piece.original_value_data_type == "text":
-                text_values.append(piece.original_value)
+        text_values.extend(
+            piece.original_value for piece in msg.message_pieces if piece.original_value_data_type == "text"
+        )
 
     return text_values