chore: linting

.hooks/check_pinned_hash_dependencies.py

@@ -10,7 +10,9 @@ def __init__(self) -> None:
         self.pinned_pattern = re.compile(r"uses:\s+([^@\s]+)@([a-f0-9]{40})")
 
         # Pattern for actions with version tags (unpinned)
-        self.unpinned_pattern = re.compile(r"uses:\s+([^@\s]+)@(v\d+(?:\.\d+)*(?:-[a-zA-Z0-9]+(?:\.\d+)*)?)")
+        self.unpinned_pattern = re.compile(
+            r"uses:\s+([^@\s]+)@(v\d+(?:\.\d+)*(?:-[a-zA-Z0-9]+(?:\.\d+)*)?)",
+        )
 
         # Pattern for all uses statements
         self.all_uses_pattern = re.compile(r"uses:\s+([^@\s]+)@([^\s\n]+)")
@@ -30,16 +32,18 @@ def format_terminal_link(self, file_path: str, line_number: int) -> str:
     def get_line_numbers(self, content: str, pattern: re.Pattern[str]) -> list[tuple[str, int]]:
         """Find matches with their line numbers."""
         matches = []
-        for i, line in enumerate(content.splitlines(), 1):
-            for match in pattern.finditer(line):
-                matches.append((match.group(0), i))
+        matches.extend(
+            (match.group(0), i)
+            for i, line in enumerate(content.splitlines(), 1)
+            for match in pattern.finditer(line)
+        )
         return matches
 
     def check_file(self, file_path: str) -> bool:
         """Check a single file for unpinned dependencies."""
         try:
             content = Path(file_path).read_text()
-        except Exception as e:
+        except (FileNotFoundError, PermissionError, IsADirectoryError, OSError) as e:
             print(f"\033[91mError reading file {file_path}: {e}\033[0m")
             return False
 
@@ -84,7 +88,9 @@ def check_file(self, file_path: str) -> bool:
             has_errors = True
             print("\033[91m[!] Completely unpinned (no SHA or version):\033[0m")
             for match, line_num in unpinned_without_hash:
-                print(f" |- {match} \033[90m({self.format_terminal_link(file_path, line_num)})\033[0m")
+                print(
+                    f" |- {match} \033[90m({self.format_terminal_link(file_path, line_num)})\033[0m",
+                )
 
         # Print summary
         total_actions = len(pinned_matches) + len(unpinned_matches) + len(unpinned_without_hash)

.hooks/generate_pr_description.py

@@ -9,14 +9,17 @@
 import asyncio
 import os
 import typing as t
+from pathlib import Path
 
 import rigging as rg
 import typer
 
-TRUNCATION_WARNING = "\n---\n**Note**: Due to the large size of this diff, some content has been truncated."
+TRUNCATION_WARNING = (
+    "\n---\n**Note**: Due to the large size of this diff, some content has been truncated."
+)
 
 
-@rg.prompt  # type: ignore
+@rg.prompt
 def generate_pr_description(diff: str) -> t.Annotated[str, rg.Ctx("markdown")]:  # type: ignore[empty-body]
     """
     Analyze the provided git diff and create a PR description in markdown format.
@@ -40,13 +43,19 @@ async def _run_git_command(args: list[str]) -> str:
     """
     # Validate git exists in PATH
     git_path = "git"  # Could use shutil.which("git") for more security
-    if not any(os.path.isfile(os.path.join(path, "git")) for path in os.environ["PATH"].split(os.pathsep)):
+    if not any(
+        Path(path).joinpath("git").is_file() for path in os.environ["PATH"].split(os.pathsep)
+    ):
         raise ValueError("Git executable not found in PATH")
 
     # Validate input parameters
     if not all(isinstance(arg, str) for arg in args):
         raise ValueError("All command arguments must be strings")
 
+    def check_return_code(return_code: int):
+        if return_code != 0:
+            raise RuntimeError(f"Git command failed: {stderr.decode()}")
+
     # Use os.execv for more secure command execution
     try:
         # nosec B603 - Input is validated
@@ -58,12 +67,15 @@ async def _run_git_command(args: list[str]) -> str:
         )
         stdout, stderr = await proc.communicate()
 
-        if proc.returncode != 0:
-            raise RuntimeError(f"Git command failed: {stderr.decode()}")
+        check_return_code(proc.returncode)
 
         return stdout.decode().strip()
-    except Exception as e:
-        raise RuntimeError(f"Failed to execute git command: {e}") from e
+    except FileNotFoundError as e:
+        raise RuntimeError("Git executable not found or invalid command") from e
+    except asyncio.SubprocessError as e:
+        raise RuntimeError("Error occurred while running the subprocess") from e
+    except UnicodeDecodeError as e:
+        raise RuntimeError("Failed to decode the output of the git command") from e
 
 
 async def get_diff(base_ref: str, source_ref: str, *, exclude: list[str] | None = None) -> str:

dreadnode/api/client.py

@@ -76,7 +76,7 @@ def _get_error_message(self, response: httpx.Response) -> str:
 
         try:
             obj = response.json()
-            return f'{response.status_code}: {obj.get("detail", json.dumps(obj))}'
+            return f"{response.status_code}: {obj.get('detail', json.dumps(obj))}"
         except Exception:  # noqa: BLE001
             return str(response.content)
 

dreadnode/artifact/merger.py

@@ -248,14 +248,14 @@ def _handle_overlaps(
             if existing_node["type"] == "dir" and new_node["type"] == "dir":
                 # Both are directories - merge them
                 self._merge_directory_nodes(
-                    cast(DirectoryNode, existing_node),
-                    cast(DirectoryNode, new_node),
+                    cast("DirectoryNode", existing_node),
+                    cast("DirectoryNode", new_node),
                 )
                 merged = True
             elif existing_node["type"] == "file" and new_node["type"] == "file":
                 # Both are files - propagate URIs and update if hash differs
-                existing_file = cast(FileNode, existing_node)
-                new_file = cast(FileNode, new_node)
+                existing_file = cast("FileNode", existing_node)
+                new_file = cast("FileNode", new_node)
 
                 if existing_file["hash"] != new_file["hash"]:
                     # Find the parent directory and update the file
@@ -333,7 +333,7 @@ def _update_file_in_tree(
                 return True
 
             if child["type"] == "dir" and self._update_file_in_tree(
-                cast(DirectoryNode, child),
+                cast("DirectoryNode", child),
                 old_file,
                 new_file,
             ):
@@ -400,7 +400,7 @@ def _build_path_and_hash_maps(
         """
         if node["type"] == "dir":
             # Add directory to path map
-            dir_node = cast(DirectoryNode, node)
+            dir_node = cast("DirectoryNode", node)
             dir_path = dir_node["dir_path"]
             path_map[dir_path] = dir_node
 
@@ -409,7 +409,7 @@ def _build_path_and_hash_maps(
                 self._build_path_and_hash_maps(child, path_map, hash_map)
         else:  # File node
             # Add file to path map
-            file_node = cast(FileNode, node)
+            file_node = cast("FileNode", node)
             file_path = file_node["final_real_path"]
             path_map[file_path] = file_node
 
@@ -448,13 +448,13 @@ def _merge_directory_nodes(self, target_dir: DirectoryNode, source_dir: Director
             if source_child["type"] == "dir":
                 self._merge_directory_child(
                     target_dir,
-                    cast(DirectoryNode, source_child),
+                    cast("DirectoryNode", source_child),
                     path_to_index,
                 )
             else:  # file
                 self._merge_file_child(
                     target_dir,
-                    cast(FileNode, source_child),
+                    cast("FileNode", source_child),
                     path_to_index,
                     hash_to_index,
                 )
@@ -474,9 +474,9 @@ def _build_indices(self, dir_node: DirectoryNode) -> tuple[dict[str, int], dict[
 
         for i, child in enumerate(dir_node["children"]):
             if child["type"] == "dir":
-                path_to_index[cast(DirectoryNode, child)["dir_path"]] = i
+                path_to_index[cast("DirectoryNode", child)["dir_path"]] = i
             else:  # file
-                file_child = cast(FileNode, child)
+                file_child = cast("FileNode", child)
                 path_to_index[file_child["final_real_path"]] = i
                 hash_to_index[file_child["hash"]] = i
 
@@ -496,7 +496,7 @@ def _merge_directory_child(
             existing_child = target_dir["children"][index]
             if existing_child["type"] == "dir":
                 self._merge_directory_nodes(
-                    cast(DirectoryNode, existing_child),
+                    cast("DirectoryNode", existing_child),
                     source_dir,
                 )
         else:
@@ -522,14 +522,14 @@ def _merge_file_child(
                 target_dir["children"][index] = source_file
             elif existing_child["type"] == "file":
                 # Same file - propagate URI if needed
-                self._propagate_uri(cast(FileNode, existing_child), source_file)
+                self._propagate_uri(cast("FileNode", existing_child), source_file)
         elif file_hash in hash_to_index:
             # Same file content exists but at different path
             index = hash_to_index[file_hash]
             existing_child = target_dir["children"][index]
             if existing_child["type"] == "file":
                 # Propagate URI if needed
-                self._propagate_uri(cast(FileNode, existing_child), source_file)
+                self._propagate_uri(cast("FileNode", existing_child), source_file)
             # Keep both files since they're at different paths
             target_dir["children"].append(source_file)
         else:
@@ -562,9 +562,9 @@ def _update_directory_hash(self, dir_node: DirectoryNode) -> str:
 
         for child in dir_node["children"]:
             if child["type"] == "file":
-                child_hashes.append(cast(FileNode, child)["hash"])
+                child_hashes.append(cast(FileNode, child)["hash"])  # noqa: TC006
             else:
-                child_hash = self._update_directory_hash(cast(DirectoryNode, child))
+                child_hash = self._update_directory_hash(cast(DirectoryNode, child))  # noqa: TC006
                 child_hashes.append(child_hash)
 
         child_hashes.sort()  # Ensure consistent hash regardless of order

dreadnode/artifact/tree_builder.py

@@ -262,7 +262,7 @@ def _build_tree_structure(
         }
         dir_structure[root_dir_path] = root_node
 
-        for file_path in file_nodes_by_path:
+        for file_path in file_nodes_by_path:  # noqa: PLC0206
             try:
                 rel_path = file_path.relative_to(base_dir)
                 parts = rel_path.parts

dreadnode/main.py

@@ -217,7 +217,7 @@ def initialize(self) -> None:
 
             try:
                 self._api.list_projects()
-            except Exception as e:  # noqa: BLE001
+            except Exception as e:
                 raise RuntimeError(
                     "Failed to authenticate with the provided server and token",
                 ) from e
@@ -371,42 +371,36 @@ class TaskDecorator(t.Protocol):
         def __call__(
             self,
             func: t.Callable[P, t.Awaitable[R]],
-        ) -> Task[P, R]:
-            ...
+        ) -> Task[P, R]: ...
 
         @t.overload
         def __call__(
             self,
             func: t.Callable[P, R],
-        ) -> Task[P, R]:
-            ...
+        ) -> Task[P, R]: ...
 
         def __call__(
             self,
             func: t.Callable[P, t.Awaitable[R]] | t.Callable[P, R],
-        ) -> Task[P, R]:
-            ...
+        ) -> Task[P, R]: ...
 
     class ScoredTaskDecorator(t.Protocol, t.Generic[R]):
         @t.overload
         def __call__(
             self,
             func: t.Callable[P, t.Awaitable[R]],
-        ) -> Task[P, R]:
-            ...
+        ) -> Task[P, R]: ...
 
         @t.overload
         def __call__(
             self,
             func: t.Callable[P, R],
-        ) -> Task[P, R]:
-            ...
+        ) -> Task[P, R]: ...
 
         def __call__(
             self,
             func: t.Callable[P, t.Awaitable[R]] | t.Callable[P, R],
-        ) -> Task[P, R]:
-            ...
+        ) -> Task[P, R]: ...
 
     @t.overload
     def task(
@@ -420,8 +414,7 @@ def task(
         log_output: bool = True,
         tags: t.Sequence[str] | None = None,
         **attributes: t.Any,
-    ) -> TaskDecorator:
-        ...
+    ) -> TaskDecorator: ...
 
     @t.overload
     def task(
@@ -435,8 +428,7 @@ def task(
         log_output: bool = True,
         tags: t.Sequence[str] | None = None,
         **attributes: t.Any,
-    ) -> ScoredTaskDecorator[R]:
-        ...
+    ) -> ScoredTaskDecorator[R]: ...
 
     def task(
         self,
@@ -514,7 +506,7 @@ def make_task(
                 tracer=self._get_tracer(),
                 name=_name,
                 attributes=_attributes,
-                func=t.cast(t.Callable[P, R], func),
+                func=t.cast("t.Callable[P, R]", func),
                 scorers=[
                     scorer
                     if isinstance(scorer, Scorer)
@@ -790,7 +782,6 @@ def log_metric(
                 Defaults to "task-or-run". If "task-or-run", the metric will be logged
                 to the current task or run, whichever is the nearest ancestor.
         """
-        ...
 
     @t.overload
     def log_metric(
@@ -822,7 +813,7 @@ def log_metric(
                 Defaults to "task-or-run". If "task-or-run", the metric will be logged
                 to the current task or run, whichever is the nearest ancestor.
         """
-        ...
+        ...  # noqa: PIE790
 
     @handle_internal_errors()
     def log_metric(

dreadnode/metric.py

@@ -83,7 +83,7 @@ class Scorer(t.Generic[T]):
     def from_callable(
         cls,
         tracer: Tracer,
-        func: ScorerCallable[T] | "Scorer[T]",
+        func: ScorerCallable[T] | "Scorer[T]",  # noqa: TC010
         *,
         name: str | None = None,
         tags: t.Sequence[str] | None = None,

dreadnode/serialization.py

@@ -76,7 +76,7 @@ def _handle_sequence(
             non_empty_schemas_found = True
 
     schema: JsonDict = {"type": "array"}
-    if obj_type != list:
+    if obj_type != list:  # noqa: E721
         schema["title"] = obj_type.__name__
         type_name_map = {tuple: "tuple", set: "set", frozenset: "set", deque: "deque"}
         schema["x-python-datatype"] = type_name_map.get(obj_type, obj_type.__name__)

dreadnode/task.py

@@ -52,8 +52,7 @@ def top_n(
         *,
         as_outputs: t.Literal[False] = False,
         reverse: bool = True,
-    ) -> "TaskSpanList[R]":
-        ...
+    ) -> "TaskSpanList[R]": ...
 
     @t.overload
     def top_n(
@@ -62,8 +61,7 @@ def top_n(
         *,
         as_outputs: t.Literal[True],
         reverse: bool = True,
-    ) -> list[R]:
-        ...
+    ) -> list[R]: ...
 
     def top_n(
         self,
@@ -85,7 +83,7 @@ def top_n(
         """
         sorted_ = self.sorted(reverse=reverse)[:n]
         return (
-            t.cast(list[R], [span.output for span in sorted_])
+            t.cast(list[R], [span.output for span in sorted_])  # noqa: TC006
             if as_outputs
             else TaskSpanList(sorted_)
         )
@@ -256,7 +254,7 @@ async def run(self, *args: P.args, **kwargs: P.kwargs) -> TaskSpan[R]:
                 for name, value in inputs_to_log.items()
             ]
 
-            output = t.cast(R | t.Awaitable[R], self.func(*args, **kwargs))
+            output = t.cast(R | t.Awaitable[R], self.func(*args, **kwargs))  # noqa: TC006
             if inspect.isawaitable(output):
                 output = await output