Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion src/loadpath/detect.py
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,7 @@ def _detect_django_root(repo_root: Path) -> str:
"web/src",
"client/src",
"ui/src",
"ui",
)

SKIP_REACT_PARTS = {
Expand All @@ -186,7 +187,15 @@ def _detect_django_root(repo_root: Path) -> str:
"starlight_help",
"e2e",
"cypress",
"graphiql",
"demo-app",
"demo",
"example",
"examples",
"legacy",
"legacy-ui",
}
SKIP_REACT_SUBSTRINGS = ("graphiql", "storybook", "docusaurus", "demo-app")


def _package_has_react(pkg: Path) -> bool:
Expand All @@ -198,6 +207,14 @@ def _package_has_react(pkg: Path) -> bool:
return "react" in deps or "react-dom" in deps


def _skip_react_tree(path: Path, repo_root: Path) -> bool:
parts = _rel_parts(path, repo_root)
if any(part in SKIP_REACT_PARTS for part in parts):
return True
joined = "/".join(parts).lower()
return any(token in joined for token in SKIP_REACT_SUBSTRINGS)


def _detect_react_root(repo_root: Path) -> str:
for candidate in PREFERRED_REACT_ROOTS:
path = repo_root / candidate
Expand All @@ -208,7 +225,7 @@ def _detect_react_root(repo_root: Path) -> str:
for pkg in repo_root.rglob("package.json"):
if _skip(pkg, repo_root):
continue
if any(part in SKIP_REACT_PARTS for part in _rel_parts(pkg, repo_root)):
if _skip_react_tree(pkg, repo_root):
continue
if not _package_has_react(pkg):
continue
Expand Down
131 changes: 101 additions & 30 deletions src/loadpath/extractors/django.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,34 @@

SERIALIZER_BASES = {"Serializer", "ModelSerializer", "HyperlinkedModelSerializer", "ListSerializer"}
FORM_BASES = {"Form", "ModelForm", "BaseForm", "BaseModelForm"}
FILTERSET_BASES = {"FilterSet"}
MODEL_BASES = {"Model"}
# Subpackages that are never the Django app name (billing/views/foo.py → billing).
APP_PACKAGE_DIRS = {
"views",
"viewsets",
"serializers",
"models",
"forms",
"filtersets",
"filters",
"tasks",
"admin",
"tests",
"templatetags",
"management",
"commands",
"migrations",
"actors",
"signals",
"receivers",
"services",
"handlers",
"api",
"endpoints",
"permissions",
"throttles",
}
ADMIN_BASES = {"ModelAdmin", "StackedInline", "TabularInline"}
CELERY_DECORATORS = {"shared_task", "task", "periodic_task"}
DRAMATIQ_DECORATORS = {"actor"}
Expand Down Expand Up @@ -134,32 +161,65 @@ def _truthy(node: ast.AST | None) -> bool:
return isinstance(node, ast.Constant) and node.value is True


def strip_url_anchors(route: str) -> str:
route = (route or "").strip()
if route.startswith("include:"):
return ""
if route.startswith("^"):
route = route[1:]
if route.endswith("$") and not route.endswith("\\$"):
route = route[:-1]
return route


def _replace_named_groups(route: str) -> str:
"""Turn `(?P<slug>(?:[\\w-]+))` into `{slug}` without choking on nested groups."""
out: list[str] = []
i = 0
n = len(route)
while i < n:
if route.startswith("(?P<", i):
name_end = route.find(">", i + 4)
if name_end != -1:
name = route[i + 4 : name_end]
k = name_end + 1
depth = 1
in_class = False
while k < n and depth:
ch = route[k]
escaped = k > 0 and route[k - 1] == "\\"
if not escaped:
if in_class:
if ch == "]":
in_class = False
elif ch == "[":
in_class = True
elif ch == "(":
depth += 1
elif ch == ")":
depth -= 1
k += 1
if depth == 0:
out.append("{" + name + "}")
i = k
continue
out.append(route[i])
i += 1
return "".join(out)


def pretty_url_pattern(route: str) -> str:
"""Turn `^$` / `(?P<slug>…)` into a graph-readable path fragment."""
return strip_url_anchors(_replace_named_groups(route or ""))


def _app_from_path(rel: str) -> str | None:
parts = Path(rel).parts
# backend/billing/models.py → billing
if "migrations" in parts:
idx = parts.index("migrations")
if idx > 0:
return parts[idx - 1]
for i, part in enumerate(parts):
if part in {
"models.py",
"views.py",
"serializers.py",
"urls.py",
"signals.py",
"signal_handlers.py",
"forms.py",
"tasks.py",
"admin.py",
"apps.py",
}:
return parts[i - 1] if i > 0 else None
if part == "management" and i > 0:
return parts[i - 1]
if len(parts) >= 2 and parts[-1].endswith(".py"):
return parts[-2]
return None
parts = list(Path(rel).parts)
if parts and parts[-1].endswith(".py"):
parts = parts[:-1]
while len(parts) > 1 and parts[-1] in APP_PACKAGE_DIRS:
parts.pop()
return parts[-1] if parts else None


def _module_qual(rel: str) -> str:
Expand Down Expand Up @@ -234,6 +294,8 @@ def visit_ClassDef(self, node: ast.ClassDef) -> None:
and not _has_base(node, {"TestCase", "SimpleTestCase", "TransactionTestCase", "LiveServerTestCase", "APITestCase"})
):
self._serializer(node, ntype=NodeType.FORM)
elif _has_base(node, FILTERSET_BASES) or node.name.endswith("FilterSet"):
self._serializer(node, ntype=NodeType.FORM, filterset=True)
elif _has_base(node, DJANGO_VIEW_BASES) or node.name.endswith(("View", "ViewSet")):
self._view(node)
elif any(b.split(".")[-1] in {"BaseCommand", "AppCommand", "LabelCommand"} for b in _bases(node)):
Expand Down Expand Up @@ -364,11 +426,15 @@ def _model(self, node: ast.ClassDef) -> None:
if extra.get("on_delete") == "CASCADE":
self.add_edge(model.id, rel_id, EdgeType.RELATES_TO, extra={"cascade": True})

def _serializer(self, node: ast.ClassDef, ntype: NodeType = NodeType.SERIALIZER) -> None:
def _serializer(
self, node: ast.ClassDef, ntype: NodeType = NodeType.SERIALIZER, *, filterset: bool = False
) -> None:
qname = f"{self.app}.{node.name}"
extra: dict = {"app": self.app}
if ntype is NodeType.FORM:
extra["django_form"] = True
extra["django_form"] = not filterset
if filterset:
extra["filterset"] = True
ser = self.add_node(ntype, node.name, qname, node.lineno, extra)
meta_model = None
meta_fields: list[str] | None = None
Expand Down Expand Up @@ -486,6 +552,10 @@ def _view(self, node: ast.ClassDef) -> None:
else f"{self.app}.{serializer_class.split('.')[-1]}"
)
self.add_edge(view.id, node_id(NodeType.SERIALIZER, ser_q), EdgeType.USES_SERIALIZER)
if extra.get("filterset") and extra["filterset"] not in {True, False}:
fs = str(extra["filterset"])
fs_q = fs if "." in fs and not fs.startswith("filter") else f"{self.app}.{fs.split('.')[-1]}"
self.add_edge(view.id, node_id(NodeType.FORM, fs_q), EdgeType.CALLS, confidence=0.9)
for perm in permissions:
pid = node_id(NodeType.PERMISSION, perm)
self.graph.nodes.append(
Expand Down Expand Up @@ -899,10 +969,11 @@ def _include_target(self, call: ast.Call) -> str | None:
def _route_identity(
self, route: str, include_mod: str | None, name: str | None, lineno: int
) -> tuple[str, str]:
"""Empty `path("", …)` must still show a label and a unique id."""
"""Empty `path("")` / `re_path(r"^$")` must still show a label and a unique id."""
stamp = f"{Path(self.rel_path).name}:{lineno}"
if route:
return route, f"{self.app}:{route}"
pretty = pretty_url_pattern(route)
if pretty:
return pretty, f"{self.app}:{route}"
if include_mod:
return f"include:{include_mod}", f"{self.app}:include:{include_mod}:{stamp}"
if name:
Expand Down
2 changes: 1 addition & 1 deletion src/loadpath/index.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
PY_SKIP = {"migrations"} # still extract migrations, just not skip
INDEX_EXTENSIONS = {".py", ".ts", ".tsx", ".js", ".jsx"}
# Bump when extractor/stitch node identity changes so incremental indexes rebuild.
INDEX_REVISION = "8"
INDEX_REVISION = "10"


def default_db_path(repo_root: Path) -> Path:
Expand Down

Large diffs are not rendered by default.

Large diffs are not rendered by default.

Loading
Loading