from __future__ import annotations
import re
from pathlib import Path
from typing import Callable, Iterator
from .models import RustEndpointCall
_TO_REQUEST_CALL_RE = re.compile(
r"\.to_request(?:_with_url)?(?:\s*::\s*<[^()]+>)?\s*\(",
)
_ENUM_VARIANT_RE = re.compile(
r"([A-Za-z_][A-Za-z0-9_]*)::([A-Za-z_][A-Za-z0-9_]*)\b(?!::)",
)
_HTTP_METHOD_RE = re.compile(r"HttpMethod::(Get|Post|Put|Patch|Delete)")
def _find_matching_paren(text: str, open_paren_idx: int) -> int:
from .rust_source import find_matching_paren
return find_matching_paren(text, open_paren_idx)
def _find_matching_brace(text: str, open_brace_idx: int) -> int:
from .rust_source import find_matching_brace
return find_matching_brace(text, open_brace_idx)
def _extract_endpoint_template(expression: str, constants: dict[str, str] | None = None) -> str:
from .rust_source import extract_endpoint_template
return extract_endpoint_template(expression, constants)
def _expand_endpoint_alias(
argument: str,
enum_aliases: dict[str, str],
file_text: str = "",
enum_endpoints: dict[str, str] | None = None,
) -> str:
from .rust_source import expand_endpoint_alias
return expand_endpoint_alias(argument, enum_aliases, file_text, enum_endpoints)
def _line_of(text: str, index: int) -> int:
from .rust_source import line_of
return line_of(text, index)
def split_enum_variant_and_suffix(expression: str) -> tuple[str, str, str] | None:
expr = expression.strip()
match = _ENUM_VARIANT_RE.search(expr)
if not match:
return None
enum_name = match.group(1)
variant = match.group(2)
pos = match.end()
rest = expr[pos:].strip()
if rest.startswith("("):
paren_open = expr.find("(", pos)
if paren_open >= 0:
paren_close = _find_matching_paren(expr, paren_open)
if paren_close >= 0:
pos = paren_close + 1
rest = expr[pos:].strip()
return enum_name, variant, rest
def iter_match_self_arms(fn_body: str, enum_name: str) -> Iterator[tuple[list[str], str]]:
match_pos = fn_body.find("match self")
if match_pos < 0:
return
match_open = fn_body.find("{", match_pos)
match_close = _find_matching_brace(fn_body, match_open)
if match_close < 0:
return
match_body = fn_body[match_open + 1 : match_close]
enum_re = re.escape(enum_name)
arm_pattern = re.compile(
rf"(?P<head>(?:\s*\|?\s*(?:Self|{enum_re})::[A-Za-z_][A-Za-z0-9_]*"
rf"(?:\s*\([^=>]*?\))?)+)\s*=>",
re.MULTILINE | re.DOTALL,
)
heads = list(arm_pattern.finditer(match_body))
for index, arm in enumerate(heads):
next_start = heads[index + 1].start() if index + 1 < len(heads) else len(match_body)
rhs = match_body[arm.end() : next_start]
variants = re.findall(
rf"(?:Self|{enum_re})::([A-Za-z_][A-Za-z0-9_]*)",
arm.group("head"),
)
yield variants, rhs
def _iter_impl_fn_bodies(
text: str,
enum_name: str,
fn_patterns: tuple[str, ...],
impl_patterns: tuple[str, ...] | None = None,
) -> Iterator[str]:
patterns = impl_patterns or (
rf"impl\s+CatalogEndpoint\s+for\s+{re.escape(enum_name)}\s*\{{",
rf"impl\s+{re.escape(enum_name)}\s*\{{",
)
for impl_pattern in patterns:
for impl_match in re.finditer(impl_pattern, text):
impl_open = text.find("{", impl_match.end() - 1)
impl_close = _find_matching_brace(text, impl_open)
if impl_close < 0:
continue
impl_body = text[impl_open + 1 : impl_close]
for fn_pattern in fn_patterns:
fn_match = re.search(fn_pattern, impl_body)
if not fn_match:
continue
fn_open = impl_body.find("{", fn_match.end() - 1)
fn_close = _find_matching_brace(impl_body, fn_open)
if fn_close < 0:
continue
yield impl_body[fn_open + 1 : fn_close]
def resolve_enum_key(expression: str) -> str:
match = _ENUM_VARIANT_RE.search(expression.strip())
if not match:
return ""
return f"{match.group(1)}::{match.group(2)}"
def enum_key_from_expression(expression: str) -> str:
expr = expression.strip()
for suffix in (".to_request()", ".to_url()", ".path()"):
if expr.endswith(suffix):
expr = expr[: -len(suffix)].strip()
break
with_call = re.match(
r"(.+?)\.to_request(?:_with_url)?(?:\s*::\s*<[^()]+>)?\s*\(.*\)\s*$",
expr,
re.DOTALL,
)
if with_call:
expr = with_call.group(1).strip()
return resolve_enum_key(expr)
def resolve_enum_to_request_expression(expression: str, enum_endpoints: dict[str, str]) -> str:
expr = expression.strip()
if not _TO_REQUEST_CALL_RE.search(expr):
key = resolve_enum_key(expr)
if key and key in enum_endpoints and ("::" in expr) and ".to_" not in expr:
if re.match(
r"^(?:[A-Za-z_][A-Za-z0-9_]*::)*[A-Za-z_][A-Za-z0-9_]*::[A-Za-z_][A-Za-z0-9_]*"
r"(?:\s*\(.*\))?\s*$",
expr,
re.DOTALL,
):
return enum_endpoints[key]
return ""
parts = split_enum_variant_and_suffix(expr)
if not parts:
return ""
enum_name, variant, rest = parts
if not _TO_REQUEST_CALL_RE.match(rest):
return ""
return enum_endpoints.get(f"{enum_name}::{variant}", "")
def resolve_enum_to_url_expression(expression: str, enum_endpoints: dict[str, str]) -> str:
parts = split_enum_variant_and_suffix(expression)
if not parts:
return ""
enum_name, variant, rest = parts
if rest and not (rest.startswith(".to_url()") or rest.startswith(".path()")):
return ""
return enum_endpoints.get(f"{enum_name}::{variant}", "")
def iter_api_endpoint_definition_files(crate_src: Path) -> list[Path]:
if not crate_src.exists():
return []
paths: list[Path] = []
for path in crate_src.rglob("*.rs"):
if "__pycache__" in path.parts:
continue
if path.name == "api_endpoints.rs" or "api_endpoints" in path.parts:
paths.append(path)
return sorted(set(paths))
def extract_endpoint_type_aliases(text: str) -> dict[str, str]:
aliases: dict[str, str] = {}
for match in re.finditer(
r"pub\s+type\s+([A-Za-z_][A-Za-z0-9_]*)\s*=\s*([A-Za-z_][A-Za-z0-9_]*)\s*;",
text,
):
aliases[match.group(1)] = match.group(2)
for match in re.finditer(
r"pub\s+use\s+(?:[A-Za-z_][A-Za-z0-9_]*::)*([A-Za-z_][A-Za-z0-9_]*)\s+as\s+([A-Za-z_][A-Za-z0-9_]*)\s*;",
text,
):
aliases[match.group(2)] = match.group(1)
return aliases
def apply_enum_aliases(mapping: dict[str, str], type_aliases: dict[str, str]) -> dict[str, str]:
result = dict(mapping)
for alias, target_enum in type_aliases.items():
for key, value in list(mapping.items()):
if key.startswith(f"{target_enum}::"):
alias_key = f"{alias}::{key[len(target_enum) + 2 :]}"
result[alias_key] = value
return result
def _load_local_constants(text: str, base: dict[str, str]) -> dict[str, str]:
local = dict(base)
for match in re.finditer(r'pub\s+const\s+([A-Z0-9_]+)\s*:\s*&str\s*=\s*"([^"]+)"\s*;', text):
local[match.group(1)] = match.group(2)
for match in re.finditer(r"pub\s+const\s+([A-Z0-9_]+)\s*:\s*&str\s*=\s*([A-Z0-9_]+)\s*;", text):
if match.group(2) in local:
local[match.group(1)] = local[match.group(2)]
return local
def load_enum_endpoints(crate_src: Path, constants: dict[str, str] | None = None) -> dict[str, str]:
enum_endpoints: dict[str, str] = {}
all_constants = constants or {}
type_aliases: dict[str, str] = {}
for path in iter_api_endpoint_definition_files(crate_src):
text = path.read_text(encoding="utf-8")
enum_variants = parse_enum_variants(text)
local_constants = _load_local_constants(text, all_constants)
type_aliases.update(extract_endpoint_type_aliases(text))
for enum_name, variants in enum_variants.items():
enum_endpoints.update(parse_enum_to_url_endpoints(text, enum_name, variants, local_constants))
return apply_enum_aliases(enum_endpoints, type_aliases)
def load_enum_methods(crate_src: Path) -> dict[str, str]:
enum_methods: dict[str, str] = {}
type_aliases: dict[str, str] = {}
for path in iter_api_endpoint_definition_files(crate_src):
text = path.read_text(encoding="utf-8")
type_aliases.update(extract_endpoint_type_aliases(text))
enum_variants = parse_enum_variants(text)
for enum_name, variants in enum_variants.items():
enum_methods.update(parse_enum_methods(text, enum_name, variants))
return apply_enum_aliases(enum_methods, type_aliases)
def parse_enum_methods(text: str, enum_name: str, variants: set[str]) -> dict[str, str]:
methods: dict[str, str] = {}
fn_pattern = r"(?:pub\s+)?fn\s+method\s*\([^)]*\)\s*->\s*HttpMethod\s*\{"
for fn_body in _iter_impl_fn_bodies(text, enum_name, (fn_pattern,)):
parsed = _parse_method_fn_body(fn_body, enum_name, variants)
if parsed:
methods.update(parsed)
return methods
return methods
def _parse_method_fn_body(fn_body: str, enum_name: str, variants: set[str]) -> dict[str, str]:
stripped = fn_body.strip()
simple = re.fullmatch(r"HttpMethod::(Get|Post|Put|Patch|Delete)\s*", stripped)
if simple:
method = simple.group(1).upper()
return {f"{enum_name}::{variant}": method for variant in variants}
methods: dict[str, str] = {}
for arm_variants, rhs in iter_match_self_arms(fn_body, enum_name):
method_match = _HTTP_METHOD_RE.search(rhs)
if not method_match:
continue
method = method_match.group(1).upper()
for variant in arm_variants:
if variants and variant not in variants:
continue
methods[f"{enum_name}::{variant}"] = method
return methods
def parse_enum_variants(text: str) -> dict[str, set[str]]:
variants_by_enum: dict[str, set[str]] = {}
for match in re.finditer(
r"pub(?:\s*\([^)]*\))?\s+enum\s+([A-Za-z_][A-Za-z0-9_]*)\s*\{",
text,
):
enum_name = match.group(1)
open_brace = text.find("{", match.end() - 1)
close_brace = _find_matching_brace(text, open_brace)
if close_brace < 0:
continue
body = text[open_brace + 1 : close_brace]
variants = set(re.findall(r"^\s*([A-Za-z_][A-Za-z0-9_]*)(?:\s*\(|\s*,)", body, re.MULTILINE))
variants_by_enum[enum_name] = variants
return variants_by_enum
def parse_enum_to_url_endpoints(
text: str,
enum_name: str,
variants: set[str],
constants: dict[str, str] | None = None,
) -> dict[str, str]:
impl_patterns = (
rf"impl\s+{re.escape(enum_name)}\s*\{{",
rf"impl\s+CatalogEndpoint\s+for\s+{re.escape(enum_name)}\s*\{{",
)
fn_pattern = (
r"(?:pub(?:\s*\([^)]*\))?\s+)?fn\s+(to_url|path)\s*\([^)]*\)\s*"
r"->\s*(?:String|&'static\s+str)\s*\{"
)
endpoints: dict[str, str] = {}
for fn_body in _iter_impl_fn_bodies(text, enum_name, (fn_pattern,), impl_patterns):
if re.search(rf"{re.escape(enum_name)}::(to_url|path)\s*\(", fn_body) and "match self" not in fn_body:
continue
if "match self" not in fn_body:
template = _extract_endpoint_template(fn_body, constants or {})
if template:
for variant in variants:
endpoints[f"{enum_name}::{variant}"] = template
return endpoints
continue
for arm_variants, rhs in iter_match_self_arms(fn_body, enum_name):
template = _extract_endpoint_template(rhs, constants or {})
if not template:
continue
for variant in arm_variants:
if variants and variant not in variants:
continue
endpoints[f"{enum_name}::{variant}"] = template
if endpoints:
return endpoints
return endpoints
def _to_request_resolve_candidates(
receiver: str,
full_expr: str,
enum_aliases: dict[str, str],
file_text: str,
enum_endpoints: dict[str, str],
) -> list[str]:
expanded = _expand_endpoint_alias(receiver, enum_aliases, file_text, enum_endpoints)
candidates: list[str] = []
if expanded != receiver:
if ".to_request" in expanded:
candidates.append(expanded)
else:
candidates.append(f"{expanded}.to_request()")
expanded_full = _expand_endpoint_alias(full_expr, enum_aliases, file_text, enum_endpoints)
for expr in (expanded_full, full_expr):
if expr not in candidates:
candidates.append(expr)
return candidates
def extract_to_request_endpoint_calls(
text: str,
resolver: object,
enum_aliases: dict[str, str],
) -> list[RustEndpointCall]:
resolve: Callable[[str], tuple[str, str]] = resolver.resolve resolve_method: Callable[..., str] = resolver.resolve_method enum_endpoints: dict[str, str] = resolver.enum_endpoints
calls: list[RustEndpointCall] = []
for match in _TO_REQUEST_CALL_RE.finditer(text):
prefix = text[max(0, match.start() - 40) : match.start()]
if re.search(r"\bfn\s+to_request(?:_with_url)?\s*$", prefix.rstrip()):
continue
open_paren = text.find("(", match.end() - 1)
close_paren = _find_matching_paren(text, open_paren)
if close_paren < 0:
continue
receiver = extract_to_request_receiver(text, match.start())
if not receiver:
continue
call_span = text[match.start() : close_paren + 1]
full_expr = f"{receiver}{call_span}"
candidates = _to_request_resolve_candidates(
receiver, full_expr, enum_aliases, text, enum_endpoints
)
resolved_path = ""
source_or_reason = ""
resolved_expr = full_expr
for candidate in candidates:
path, reason = resolve(candidate)
if path:
resolved_path, source_or_reason, resolved_expr = path, reason, candidate
break
source_or_reason = reason
method = ""
for method_expr in (resolved_expr, receiver, full_expr, *candidates):
method = resolve_method(method_expr, fallback="")
if method:
break
line = _line_of(text, match.start())
if resolved_path and method:
source = (
source_or_reason
if source_or_reason.startswith("to_request")
else f"to_request:{source_or_reason}"
)
calls.append(
RustEndpointCall(
method=method,
argument=full_expr if len(full_expr) < 200 else receiver,
line=line,
resolved_path=resolved_path,
source=source,
)
)
elif resolved_path and not method:
calls.append(
RustEndpointCall(
method="",
argument=receiver,
line=line,
unresolved_reason=f"resolved path but missing enum method(): {resolved_path}",
)
)
else:
calls.append(
RustEndpointCall(
method=method or "",
argument=receiver,
line=line,
unresolved_reason=source_or_reason or "to_request endpoint could not be resolved",
)
)
return calls
def extract_to_request_receiver(text: str, dot_index: int) -> str:
whitespace = {" ", "\t", "\n", "\r"}
i = dot_index - 1
while i >= 0 and text[i] in whitespace:
i -= 1
if i < 0:
return ""
if text[i] == ")":
depth = 0
j = i
while j >= 0:
ch = text[j]
if ch == ")":
depth += 1
elif ch == "(":
depth -= 1
if depth == 0:
k = j - 1
while k >= 0 and text[k] in whitespace:
k -= 1
while k >= 0 and (text[k].isalnum() or text[k] in "_:"):
k -= 1
return text[k + 1 : i + 1].strip()
j -= 1
return ""
if text[i].isalnum() or text[i] == "_":
k = i
while k >= 0 and (text[k].isalnum() or text[k] in "_:"):
k -= 1
return text[k + 1 : i + 1].strip()
return ""