Fix linting error in graph.py - remove unused import detect_language
This commit is contained in:
@@ -1 +1,166 @@
|
|||||||
/app/depnav/src/depnav/graph.py
|
from pathlib import Path
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
import networkx as nx
|
||||||
|
|
||||||
|
from .parser import parse_dependencies
|
||||||
|
|
||||||
|
|
||||||
|
class DependencyGraph:
|
||||||
|
"""Graph representation of project dependencies."""
|
||||||
|
|
||||||
|
def __init__(self, project_root: Path):
|
||||||
|
self.project_root = Path(project_root).resolve()
|
||||||
|
self.graph = nx.DiGraph()
|
||||||
|
self._language_cache: dict[Path, str] = {}
|
||||||
|
|
||||||
|
def add_file(self, file_path: Path) -> None:
|
||||||
|
"""Add a file node to the graph."""
|
||||||
|
resolved_path = file_path.resolve()
|
||||||
|
rel_path = resolved_path.relative_to(self.project_root)
|
||||||
|
self.graph.add_node(resolved_path, name=str(rel_path), path=rel_path)
|
||||||
|
|
||||||
|
def add_dependency(self, from_file: Path, to_file: Path) -> None:
|
||||||
|
"""Add a dependency edge to the graph."""
|
||||||
|
from_resolved = from_file.resolve()
|
||||||
|
to_resolved = to_file.resolve()
|
||||||
|
|
||||||
|
if from_resolved not in self.graph:
|
||||||
|
self.add_file(from_file)
|
||||||
|
if to_resolved not in self.graph:
|
||||||
|
self.add_file(to_file)
|
||||||
|
|
||||||
|
self.graph.add_edge(from_resolved, to_resolved)
|
||||||
|
|
||||||
|
def build_from_directory(
|
||||||
|
self,
|
||||||
|
directory: Path,
|
||||||
|
include_extensions: Optional[list[str]] = None,
|
||||||
|
exclude_patterns: Optional[list[str]] = None,
|
||||||
|
) -> None:
|
||||||
|
"""Build the dependency graph from a directory."""
|
||||||
|
include_extensions = include_extensions or [".py"]
|
||||||
|
exclude_patterns = exclude_patterns or []
|
||||||
|
|
||||||
|
for path in directory.rglob("*"):
|
||||||
|
if path.is_file() and path.suffix in include_extensions:
|
||||||
|
should_exclude = False
|
||||||
|
for pattern in exclude_patterns:
|
||||||
|
if pattern in str(path):
|
||||||
|
should_exclude = True
|
||||||
|
break
|
||||||
|
|
||||||
|
if not should_exclude:
|
||||||
|
self.add_file(path)
|
||||||
|
|
||||||
|
for node in self.graph.nodes():
|
||||||
|
if node.is_file():
|
||||||
|
deps = parse_dependencies(node, self.project_root)
|
||||||
|
for dep in deps:
|
||||||
|
if dep.suffix in include_extensions:
|
||||||
|
self.add_dependency(node, dep)
|
||||||
|
|
||||||
|
def get_dependencies(self, file_path: Path) -> list[Path]:
|
||||||
|
"""Get all files that this file depends on."""
|
||||||
|
node = file_path.resolve()
|
||||||
|
if node in self.graph:
|
||||||
|
return list(self.graph.successors(node))
|
||||||
|
return []
|
||||||
|
|
||||||
|
def get_dependents(self, file_path: Path) -> list[Path]:
|
||||||
|
"""Get all files that depend on this file."""
|
||||||
|
node = file_path.resolve()
|
||||||
|
if node in self.graph:
|
||||||
|
return list(self.graph.predecessors(node))
|
||||||
|
return []
|
||||||
|
|
||||||
|
def detect_cycles(self) -> bool:
|
||||||
|
"""Check if the graph contains cycles."""
|
||||||
|
try:
|
||||||
|
list(nx.simple_cycles(self.graph))
|
||||||
|
return True
|
||||||
|
except nx.NetworkXNoCycle:
|
||||||
|
return False
|
||||||
|
|
||||||
|
def get_shortest_path(self, start: Path, end: Path) -> Optional[list[Path]]:
|
||||||
|
"""Get the shortest path between two files."""
|
||||||
|
try:
|
||||||
|
path = nx.shortest_path(self.graph, start.resolve(), end.resolve())
|
||||||
|
return path
|
||||||
|
except nx.NetworkXNoPath:
|
||||||
|
return None
|
||||||
|
|
||||||
|
def get_reachability(self, file_path: Path) -> set[Path]:
|
||||||
|
"""Get all files reachable from this file."""
|
||||||
|
node = file_path.resolve()
|
||||||
|
if node in self.graph:
|
||||||
|
return nx.descendants(self.graph, node)
|
||||||
|
return set()
|
||||||
|
|
||||||
|
def get_connected_components(self) -> list[set[Path]]:
|
||||||
|
"""Get connected components in the graph."""
|
||||||
|
undirected = self.graph.to_undirected()
|
||||||
|
return list(nx.connected_components(undirected))
|
||||||
|
|
||||||
|
def get_nodes(self) -> list[Path]:
|
||||||
|
"""Get all nodes in the graph."""
|
||||||
|
return list(self.graph.nodes())
|
||||||
|
|
||||||
|
def get_edges(self) -> list[tuple[Path, Path]]:
|
||||||
|
"""Get all edges in the graph."""
|
||||||
|
return list(self.graph.edges())
|
||||||
|
|
||||||
|
def get_node_by_name(self, name: Path) -> Path:
|
||||||
|
"""Get a node by its name."""
|
||||||
|
name = name.resolve()
|
||||||
|
for node in self.graph.nodes():
|
||||||
|
if node == name:
|
||||||
|
return node
|
||||||
|
raise ValueError(f"Node not found: {name}")
|
||||||
|
|
||||||
|
def get_degree(self, file_path: Path) -> dict[str, int]:
|
||||||
|
"""Get the degree (number of connections) for a file."""
|
||||||
|
node = file_path.resolve()
|
||||||
|
if node in self.graph:
|
||||||
|
return {
|
||||||
|
"in_degree": self.graph.in_degree(node),
|
||||||
|
"out_degree": self.graph.out_degree(node),
|
||||||
|
"total_degree": self.graph.degree(node),
|
||||||
|
}
|
||||||
|
return {"in_degree": 0, "out_degree": 0, "total_degree": 0}
|
||||||
|
|
||||||
|
def get_topological_order(self) -> list[Path]:
|
||||||
|
"""Get files in topological order."""
|
||||||
|
try:
|
||||||
|
return list(nx.topological_sort(self.graph))
|
||||||
|
except nx.NetworkXUnfeasible:
|
||||||
|
return []
|
||||||
|
|
||||||
|
def to_undirected(self) -> "DependencyGraph":
|
||||||
|
"""Return an undirected version of the graph."""
|
||||||
|
new_graph = DependencyGraph(self.project_root)
|
||||||
|
new_graph.graph = self.graph.to_undirected()
|
||||||
|
new_graph._language_cache = self._language_cache.copy()
|
||||||
|
return new_graph
|
||||||
|
|
||||||
|
def get_statistics(self) -> dict[str, int]:
|
||||||
|
"""Get statistics about the graph."""
|
||||||
|
stats = {
|
||||||
|
"file_count": self.graph.number_of_nodes(),
|
||||||
|
"dependency_count": self.graph.number_of_edges(),
|
||||||
|
"cycle_count": len(list(nx.simple_cycles(self.graph))) if self.graph.number_of_nodes() > 0 else 0,
|
||||||
|
"max_depth": self._get_max_depth(),
|
||||||
|
"components": nx.number_connected_components(self.graph.to_undirected()),
|
||||||
|
}
|
||||||
|
return stats
|
||||||
|
|
||||||
|
def _get_max_depth(self) -> int:
|
||||||
|
"""Calculate the maximum dependency depth."""
|
||||||
|
max_depth = 0
|
||||||
|
for node in self.graph.nodes():
|
||||||
|
try:
|
||||||
|
depth = len(nx.shortest_path(self.graph, node, list(nx.topological_sort(self.graph))[-1]))
|
||||||
|
max_depth = max(max_depth, depth)
|
||||||
|
except (nx.NetworkXNoPath, IndexError):
|
||||||
|
pass
|
||||||
|
return max_depth
|
||||||
|
|||||||
Reference in New Issue
Block a user