Repository navigation
Expand file tree
/
Copy pathpath_utils.py
More file actions
659 lines (561 loc) · 26.6 KB
/
Copy pathpath_utils.py
File metadata and controls
659 lines (561 loc) · 26.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
import hashlib
import os
import re
from collections.abc import Callable, Iterable, Iterator, Mapping
from dataclasses import dataclass
from enum import StrEnum
from functools import lru_cache
from pathlib import Path
from pathspec import PathSpec
from .. import constants as cs
_PROJECT_NAME_INVALID_CHARS = re.compile(r"[^A-Za-z0-9_-]+")
_PROJECT_NAME_FALLBACK_BASE = "repo"
def derive_project_name(repo_path: Path) -> str:
resolved = repo_path.resolve()
digest = hashlib.sha256(str(resolved).encode("utf-8")).hexdigest()[
: cs.PROJECT_NAME_DIGEST_LEN
]
# Strip `-` as well as `_`: a name like `мой-repo` would otherwise start
# with `-`, and so would every qualified name under it, which Click then
# reads as an option wherever a command takes one (issue #2638).
base = _PROJECT_NAME_INVALID_CHARS.sub("_", resolved.name).strip("_-")
if not base:
base = _PROJECT_NAME_FALLBACK_BASE
return f"{base}{cs.PROJECT_NAME_DIGEST_MARKER}{digest}"
def resolve_repo_path(repo_path: str | None, target_default: str) -> Path:
if repo_path:
return Path(repo_path).resolve()
if target_default and target_default != ".":
return Path(target_default).resolve()
return Path.cwd().resolve()
def unwritable_output_reason(path: Path) -> str | None:
"""Why no file can be written at `path`, or None when one can.
Run before the graph is read or indexed, so a mistyped `-o` fails at once
instead of after the work it was meant to save (issue #2410). Missing
parent directories are fine: the writer creates them.
"""
if path.is_dir():
return cs.CLI_ERR_OUTPUT_IS_DIR.format(path=path)
if path.exists():
if os.access(path, os.W_OK):
return None
return cs.CLI_ERR_OUTPUT_NOT_WRITABLE.format(target=path)
ancestor = next((parent for parent in path.parents if parent.exists()), None)
if ancestor is None:
return None
if not ancestor.is_dir():
return cs.CLI_ERR_OUTPUT_PARENT_NOT_DIR.format(parent=ancestor)
if not os.access(ancestor, os.W_OK | os.X_OK):
return cs.CLI_ERR_OUTPUT_NOT_WRITABLE.format(target=ancestor)
return None
@lru_cache(maxsize=4096)
def cached_relative_path(file_path: Path, repo_path: Path) -> Path:
return file_path.relative_to(repo_path)
@lru_cache(maxsize=4096)
def cached_resolve_posix(file_path: Path) -> str:
return file_path.resolve().as_posix()
@lru_cache(maxsize=4096)
def cached_file_identity_posix(file_path: Path) -> str:
"""Absolute POSIX identity that survives the file's own deletion.
``resolve()`` dereferences a leaf symlink to its target, but once the link
is deleted that target can no longer be recovered, so a node keyed on it can
never be matched for deletion and leaks (GHSA-85gg aside, issue #1154).
Resolving only the parent (which outlives the leaf) and keeping the leaf
name yields a key that ingestion and deletion agree on, and one that stays
inside the repository for a link whose target points outside it.
"""
return (file_path.parent.resolve() / file_path.name).as_posix()
# #495: .cgrignore lines and --exclude values are interpreted with
# .gitignore (gitwildmatch) semantics: bare names match at any depth (as
# before), and globs / anchoring / dir-only trailing slash now work. The
# spec is compiled once per pattern set (frozensets are hashable).
@lru_cache(maxsize=64)
def compiled_ignore_spec(patterns: frozenset[str]) -> PathSpec:
return PathSpec.from_lines(cs.GITWILDMATCH_STYLE, sorted(patterns))
def matches_ignore_patterns(rel_path_str: str, patterns: frozenset[str]) -> bool:
return compiled_ignore_spec(patterns).match_file(rel_path_str)
_GLOB_MAGIC = re.compile(r"[*?\[]")
def unignore_names_this_file(rel_path_str: str, patterns: frozenset[str]) -> bool:
"""Whether some unignore pattern names THIS FILE rather than a container.
`matches_ignore_patterns` cannot answer this: gitwildmatch matches
`build/js/jquery.min.js` against the directory pattern `build` just as
readily as against the file's own path, so a directory-level `!` would
otherwise rescue a bundle inside it.
This is the "what counts as exact" decision issue #1637 left open. A
pattern qualifies when, with any trailing slash removed, it equals the
file's path or its bare filename -- so `!docs/js/jquery.min.js` and
`!jquery.min.js` both rescue, while `!docs`, `!docs/js` and `!docs/**` do
not. Globs are deliberately excluded: `!docs/**` reads as "rescue this
subtree", which is the directory-level intent the split preserves.
"""
filename = rel_path_str.rsplit(cs.SEPARATOR_SLASH, 1)[-1]
for pattern in patterns:
candidate = pattern.strip().rstrip(cs.SEPARATOR_SLASH)
if not candidate or _GLOB_MAGIC.search(candidate):
continue
if candidate.lstrip(cs.SEPARATOR_SLASH) in (rel_path_str, filename):
return True
return False
def unignore_could_match_within(pattern: str, rel_dir: str) -> bool:
# Dir-pruning guard: keep a pruned-by-default directory when an
# unignore pattern could match it or anything beneath it.
if "/" not in pattern.rstrip("/"):
# slash-free patterns are unanchored: they can match at any depth.
return True
head, *glob_rest = _GLOB_MAGIC.split(pattern, 1)
if glob_rest:
# the glob may complete the trailing segment; keep whole segments.
head = head.rsplit("/", 1)[0]
head = head.strip("/")
return (
not head
or head == rel_dir
or head.startswith(f"{rel_dir}/")
or rel_dir.startswith(f"{head}/")
)
def should_keep_dir(
dirname: str,
dir_prefix: str,
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
) -> bool:
"""Whether a repository walk descends into this directory.
The one predicate every whole-repo walk prunes with, so a sweep that
reads files the indexer skipped (or skips files the indexer holds)
cannot happen by construction (issue #1088).
"""
rel_dir = f"{dir_prefix}{dirname}"
# an explicit exclude can never be rescued by unignore (excludes win
# at the file level too), so prune the subtree outright.
if exclude_paths and matches_ignore_patterns(f"{rel_dir}/", exclude_paths):
return False
if dirname not in cs.IGNORE_PATTERNS:
return True
# Cargo's src/bin/ holds first-party binaries, not build output;
# mirrors has_ignored_dir_part.
if (
dirname == cs.DIR_BIN
and dir_prefix.rstrip(cs.SEPARATOR_SLASH).rsplit(cs.SEPARATOR_SLASH, 1)[-1]
== cs.DIR_SRC
):
return True
return bool(
unignore_paths
and any(unignore_could_match_within(u, rel_dir) for u in unignore_paths)
)
def is_ignored_filename(name: str) -> bool:
"""Whether a filename is a machine-generated artefact, by its ending.
The single definition of the `IGNORE_SUFFIXES` rule, shared by the
repository walk and the real-time watcher. It tests the whole filename
rather than `Path.suffix` because the list holds endings that are not
pathlib suffixes: `Path("jquery.min.js").suffix` is ".js" and
`Path("notes.py~").suffix` is ".py~", so a `Path.suffix` membership test
matched neither, and the two consumers disagreed about which files belong
in the graph -- `~` was live for the watcher and dead for the indexer
(issue #1636).
"""
return name.endswith(cs.IGNORE_FILENAME_ENDINGS)
def is_unconditionally_ignored_filename(name: str) -> bool:
"""Whether a filename is ignored even against an explicit unignore.
The stricter half of `is_ignored_filename`: compiled output and editor
droppings, which no configuration should resurrect. The rest of the list
is text a parser can read, so an explicit `!` line rescues it instead
(issue #1637). Both walk predicates must ask this question the same way,
or the indexer and the watcher disagree about which files are in the graph.
"""
return name.endswith(cs.UNCONDITIONAL_IGNORE_FILENAME_ENDINGS)
def should_skip_path(
path: Path,
repo_path: Path,
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
is_file: bool | None = None,
) -> bool:
_is_file = path.is_file() if is_file is None else is_file
# Ahead of every exclude/unignore check for the UNCONDITIONAL half only:
# compiled output is not source in any configuration, so rescuing the
# DIRECTORY it sits in must not drag it back in. The rescuable half is
# tested after the unignore below, so an explicit `!` line can win.
if _is_file and is_unconditionally_ignored_filename(path.name):
return True
# Containment below is lexical, so a symlink whose target escapes the root
# would pass it and let a repo-scoped sweep read or overwrite outside files
# (GHSA-85gg-2gfq-q95m). Resolve first, mirroring validate_project_path
# (decorators.py) and absolute_path_within_project_root (this module).
try:
path.resolve().relative_to(repo_path.resolve())
except ValueError:
return True
rel_path = cached_relative_path(path, repo_path)
rel_path_str = rel_path.as_posix()
# a trailing slash marks the path as a directory for dir-only patterns.
match_path = rel_path_str if _is_file else f"{rel_path_str}/"
if exclude_paths and matches_ignore_patterns(match_path, exclude_paths):
return True
# The rescuable half, decided BEFORE the general unignore below: that check
# returns False for a directory-level `!` too, since gitwildmatch matches
# `build/js/jquery.min.js` against the pattern `build`. Requiring a pattern
# that names the FILE is what keeps `!build/` from resurrecting a bundle
# inside it while `!build/js/jquery.min.js` rescues it (issue #1637).
if _is_file and is_ignored_filename(path.name):
return not (
unignore_paths and unignore_names_this_file(rel_path_str, unignore_paths)
)
# unignore rescues only built-in ignores, never explicit user excludes.
if unignore_paths and matches_ignore_patterns(match_path, unignore_paths):
return False
if (
not _is_file
and unignore_paths
and any(unignore_could_match_within(u, rel_path_str) for u in unignore_paths)
):
# structure traversal must descend into a built-in-ignored dir when
# an unignore pattern can match beneath it (mirrors _should_keep_dir),
# or rescued files get no Folder/Package ancestry in the graph.
return False
dir_parts = rel_path.parent.parts if _is_file else rel_path.parts
return has_ignored_dir_part(dir_parts)
def has_ignored_dir_part(dir_parts: tuple[str, ...]) -> bool:
# `bin` is a build-output ignore (dotnet's <proj>/bin, repo-root bin/)
# EXCEPT directly under src/: Cargo's multi-binary layout puts
# first-party binaries in src/bin/, where build systems never emit.
for index, part in enumerate(dir_parts):
if part not in cs.IGNORE_PATTERNS:
continue
if part == cs.DIR_BIN and index > 0 and dir_parts[index - 1] == cs.DIR_SRC:
continue
return True
return False
def should_skip_rel_file(
rel_path_str: str,
dir_parts: tuple[str, ...],
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
) -> bool:
# The filename comes from `rel_path_str` rather than a caller-supplied
# suffix: every caller derived that suffix with a last-dot split, which
# cannot see a compound ending like ".min.js" no matter what this function
# then does with it (issue #1636). First, matching `should_skip_path`; the
# two must agree on precedence as well as on the rule.
filename = rel_path_str.rsplit(cs.SEPARATOR_SLASH, 1)[-1]
if is_unconditionally_ignored_filename(filename):
return True
if exclude_paths and matches_ignore_patterns(rel_path_str, exclude_paths):
return True
# Same position and rule as `should_skip_path`: the rescuable half needs a
# pattern naming the FILE, decided before the general unignore below, or a
# directory-level `!` would rescue it there (#1637). The two predicates
# must agree on precedence, not merely on the ending rule.
if is_ignored_filename(filename):
return not (
unignore_paths and unignore_names_this_file(rel_path_str, unignore_paths)
)
# unignore rescues only built-in ignores, never explicit user excludes.
if unignore_paths and matches_ignore_patterns(rel_path_str, unignore_paths):
return False
return has_ignored_dir_part(dir_parts)
# Longest first, so `.d.ts` is tried before the `.ts` it ends with. Derived
# from the language set rather than restated, so an extension added there
# cannot silently fall through to the single-suffix rule below (issue #1720).
_MODULE_EXTS_LONGEST_FIRST: tuple[str, ...] = tuple(
sorted(cs.JS_TS_MODULE_EXTENSIONS, key=str.__len__, reverse=True)
)
def module_stem(filename: str) -> str:
"""The filename with its MODULE extension removed, compound ones included.
``Path.with_suffix("")`` strips one dot-segment, which is right for `.py`
and wrong for `.d.ts`: TypeScript declaration files carry a two-segment
extension that the language treats as a unit, and every other part of cgr
already does too -- ``JS_TS_MODULE_EXTENSIONS`` lists `.d.ts`, and the
JS/TS resolver looks up `pkg/index.d.ts` as the `pkg` entry point. Only the
qn derivation disagreed, storing it as `proj.pkg.index.d`, a name no
importer ever asks for, so its definitions were unreachable (issue #1720).
Extensions outside the language set keep the single-suffix behaviour:
`archive.tar.gz` is not a module named `archive`, and a `.d` in some other
language is not a declaration marker.
"""
for ext in _MODULE_EXTS_LONGEST_FIRST:
if filename.endswith(ext) and len(filename) > len(ext):
return filename[: -len(ext)]
return Path(filename).stem
def module_extension(filename: str) -> str:
"""The extension `module_stem` removes, dot included (`.py`, `.d.ts`)."""
return filename[len(module_stem(filename)) :]
_DECLARATION_EXTS: tuple[str, ...] = tuple(
ext
for ext in _MODULE_EXTS_LONGEST_FIRST
if ext.startswith(cs.DECLARATION_EXT_PREFIX)
)
_IMPLEMENTATION_EXTS: tuple[str, ...] = tuple(
ext for ext in _MODULE_EXTS_LONGEST_FIRST if ext not in _DECLARATION_EXTS
)
def declaration_extension(filename: str) -> str | None:
"""The TYPE-ONLY module extension this file carries, if any.
Partitioned from the one language set rather than listed again, so a
declaration form added to ``JS_TS_MODULE_EXTENSIONS`` is classified here
without a second edit.
"""
for ext in _DECLARATION_EXTS:
if filename.endswith(ext) and len(filename) > len(ext):
return ext
return None
def has_implementation_sibling(
path: Path,
repo_path: Path,
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
) -> bool:
"""Will a same-stem NON-declaration module actually be INDEXED?
Asked of the filesystem rather than of the parse registry deliberately.
The registry answers "has one been seen yet", which depends on walk order;
the disk answers "does one exist", which does not. The distinction is the
whole point -- a `.d.ts` sorts before its `.ts` in an ascending walk, so a
registry-based tie-break hands the shared name to the stub every time
(issue #1720, and the regression that closed PR #1721).
But existence alone is too permissive, and the gap is not hypothetical: an
excluded `foo.ts` is on disk and will never be a node, so yielding
`proj.foo` to it leaves NOTHING owning the name imports of `./foo` resolve
to -- #1720 reintroduced by its own fix, under a configuration the first
version of this predicate could not see (Greptile P1 on PR #1728).
So candidates are filtered through `should_skip_path`, the same predicate
the repository walk prunes with. Sharing it is what stops this from
disagreeing with the indexer about which files exist (issue #1088); asking
the question a second way here is how the two would drift apart.
"""
stem = module_stem(path.name)
for ext in _IMPLEMENTATION_EXTS:
candidate = path.parent / f"{stem}{ext}"
if not candidate.is_file():
continue
if should_skip_path(
candidate,
repo_path,
exclude_paths=exclude_paths,
unignore_paths=unignore_paths,
is_file=True,
):
continue
return True
return False
def base_module_qn(rel_path: Path, project_name: str) -> str:
"""The module qualified name for a file, BEFORE collision disambiguation.
Shared by the tree-sitter indexer (``DefinitionProcessor.process_file``)
and the C++ module-qn map (``cpp_frontend.qn``), which must agree
byte-for-byte: the whole graph keys on these names, and a disagreement
silently splits one module into two nodes rather than raising (#1025).
``__init__.py`` and ``mod.rs`` name their PACKAGE rather than themselves,
so they drop their own filename segment.
"""
if rel_path.name in (cs.INIT_PY, cs.MOD_RS):
parts = rel_path.parent.parts
else:
parts = (*rel_path.parent.parts, module_stem(rel_path.name))
return cs.SEPARATOR_DOT.join([project_name, *parts])
def _walk_dir_keys(
dirpath: str, repo_prefix_len: int
) -> tuple[tuple[str, ...], str, str]:
"""Derive a walked directory's ``(parts, cache_key, prefix)``.
Split out of ``walk_eligible_files`` so the walk body reads as filtering
alone. The repository root is the special case: it has no relative path,
and its cache key is ``ROOT_DIR_KEY`` rather than the empty string.
The relative directory itself is deliberately not returned: every caller
needs it only as the ``dir_prefix`` built here, and handing back both
invites the two to be derived independently.
"""
if len(dirpath) < repo_prefix_len:
return (), cs.ROOT_DIR_KEY, ""
rel_dir = dirpath[repo_prefix_len:].replace(os.sep, "/")
dir_parts = tuple(rel_dir.split("/")) if rel_dir else ()
return (
dir_parts,
rel_dir or cs.ROOT_DIR_KEY,
f"{rel_dir}/" if rel_dir else "",
)
def walk_eligible_files(
repo_path: Path,
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
on_dir: Callable[[str, str], None] | None = None,
) -> Iterator[tuple[str, str, str]]:
"""Yield ``(dirpath, filename, rel_path)`` for every indexable file, in order.
The single definition of the repository walk. Both the tree-sitter indexer
(``GraphUpdater._collect_eligible_files``) and the C++ module-qn map
(``cpp_frontend.qn.build_module_qn_map``) consume it, because the module-qn
disambiguation rule hands the base qn to whichever file is seen FIRST and
appends an extension to the loser -- so the two must agree on ORDER, not
merely on which files are eligible (issues #1025, #1099).
They previously kept separate copies of this loop that shared only the
filter predicates. Nothing forced the two orderings to stay equal, and a
set-based parity test cannot see the difference.
``on_dir`` receives ``(dir_key, dirpath)`` per visited directory, for the
indexer's mtime bookkeeping; it must not mutate the walk.
"""
repo_str = str(repo_path)
# A repo path that already ends in a separator (the filesystem root, "/")
# must not have another counted, or the slice below eats the first
# character of every child -- "/tmp" became "mp" (CodeRabbit, PR #1511).
repo_prefix_len = len(repo_str) + (0 if repo_str.endswith(os.sep) else 1)
state_filenames = cs.CGR_STATE_FILENAMES
for dirpath, dirnames, filenames in os.walk(repo_str):
dir_parts, dir_key, dir_prefix = _walk_dir_keys(dirpath, repo_prefix_len)
if on_dir is not None:
on_dir(dir_key, dirpath)
dirnames[:] = sorted(
d
for d in dirnames
if should_keep_dir(d, dir_prefix, exclude_paths, unignore_paths)
)
for fname in sorted(filenames):
if fname in state_filenames:
continue
rel_path_str = f"{dir_prefix}{fname}"
if not should_skip_rel_file(
rel_path_str,
dir_parts,
exclude_paths=exclude_paths,
unignore_paths=unignore_paths,
):
yield dirpath, fname, rel_path_str
def is_walked_dir(
dir_parts: tuple[str, ...],
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
) -> bool:
"""Whether `walk_eligible_files` descends into this repo-relative directory.
The walk prunes one level at a time, so a directory is reached only when
every ancestor passes `should_keep_dir` too.
"""
prefix = ""
for part in dir_parts:
if not should_keep_dir(part, prefix, exclude_paths, unignore_paths):
return False
prefix = f"{prefix}{part}{cs.SEPARATOR_SLASH}"
return True
def is_eligible_rel_file(
rel_path_str: str,
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
) -> bool:
"""Whether `walk_eligible_files` would yield this repo-relative file.
For consumers that see one path at a time, such as the real-time watcher.
Restating the rule there let the two disagree: the watcher checked
directory names against the built-in ignores only, so it dropped every
change under a directory the run had unignored, and followed changes
under one the user had excluded.
"""
*dirs, filename = rel_path_str.split(cs.SEPARATOR_SLASH)
if filename in cs.CGR_STATE_FILENAMES:
return False
dir_parts = tuple(dirs)
if not is_walked_dir(dir_parts, exclude_paths, unignore_paths):
return False
return not should_skip_rel_file(
rel_path_str,
dir_parts,
exclude_paths=exclude_paths,
unignore_paths=unignore_paths,
)
def project_roots_from_rows(
rows: Iterable[Mapping[str, object]],
) -> dict[str, str | None]:
"""Build {project_name: root_path} from CYPHER_LIST_PROJECTS rows."""
roots: dict[str, str | None] = {}
for row in rows:
name = row.get("name")
if not isinstance(name, str):
continue
root = row.get("root_path")
roots[name] = root if isinstance(root, str) else None
return roots
def project_root_for_qualified_name(
qualified_name: str, roots: dict[str, str | None]
) -> Path | None:
matches = [
name
for name in roots
if qualified_name == name or qualified_name.startswith(name + ".")
]
if not matches:
return None
owner = matches[0]
for name in matches[1:]:
if len(name) > len(owner):
owner = name
root = roots[owner]
return Path(root).resolve() if root is not None else None
def absolute_path_within_project_root(
qualified_name: str, absolute_path: str, roots: dict[str, str | None]
) -> bool:
"""A stored absolute path may only be read from inside its own project's
indexed root; projects with no recorded root (legacy graphs) stay
readable (issue #425). Project names may contain dots, so the owning
project is the longest known name prefixing the qualified name. The
resolve() calls are load-bearing: containment is checked lexically, so
an unresolved ``..`` segment or symlink would escape the root."""
root = project_root_for_qualified_name(qualified_name, roots)
if root is None:
return True
return Path(absolute_path).resolve().is_relative_to(root)
class SourceMiss(StrEnum):
"""Why a node's source file could not be located."""
STALE_ROOT = "stale_root"
OUTSIDE_ROOT = "outside_root"
MISSING = "missing"
@dataclass(frozen=True)
class SourceLocation:
"""A node's source file, or why there is none. `project` and `root` name
the owning project and its recorded root when one is known, so a stale
root can be reported as such rather than as a missing file."""
path: Path | None
miss: SourceMiss | None = None
project: str | None = None
root: Path | None = None
def _owning_project(qualified_name: str, roots: dict[str, str | None]) -> str | None:
owner = None
for name in roots:
if (qualified_name == name or qualified_name.startswith(name + ".")) and (
owner is None or len(name) > len(owner)
):
owner = name
return owner
def locate_node_source(
qualified_name: str,
absolute_path: str | None,
relative_path: str | Path,
roots: dict[str, str | None],
unowned_root: Path | None,
) -> SourceLocation:
"""The one fallback sequence both source readers use (issue #2123).
The recorded absolute path is authoritative when it sits inside the
owning project's root and is a file. Otherwise the relative path is
joined to that root, or to `unowned_root` for a node no registered
project owns; with no root at all the relative path is used as given,
the legacy behaviour. A recorded root that is no longer a directory is
reported as stale: the repository moved and must be re-indexed, and
falling back to another root would read another project's file (#1881).
"""
project = _owning_project(qualified_name, roots)
recorded = roots.get(project) if project is not None else None
owner_root = Path(recorded).resolve() if recorded is not None else None
if (
absolute_path
and absolute_path_within_project_root(qualified_name, absolute_path, roots)
and Path(absolute_path).is_file()
):
return SourceLocation(Path(absolute_path), project=project, root=owner_root)
if recorded is not None and owner_root is not None and not owner_root.is_dir():
# Reported as recorded: resolving a vanished path rewrites it through
# whatever the host maps it to, which is not what the user indexed.
return SourceLocation(None, SourceMiss.STALE_ROOT, project, Path(recorded))
base = owner_root or unowned_root
# With no root the path is used exactly as the caller validated it.
if base is None:
candidate = (
Path(relative_path) if isinstance(relative_path, str) else relative_path
)
else:
candidate = (base / relative_path).resolve()
if base is not None and not candidate.is_relative_to(base):
return SourceLocation(None, SourceMiss.OUTSIDE_ROOT, project, owner_root)
if not candidate.is_file():
return SourceLocation(None, SourceMiss.MISSING, project, owner_root)
return SourceLocation(candidate, project=project, root=owner_root)