Skip to content

Instantly share code, notes, and snippets.

@stephanGarland
Created August 11, 2026 20:20
Show Gist options
  • Select an option

  • Save stephanGarland/837ac23212130857dbff73a55199dbea to your computer and use it in GitHub Desktop.

Select an option

Save stephanGarland/837ac23212130857dbff73a55199dbea to your computer and use it in GitHub Desktop.
PyREPL - bad name, useful tool - allows you to easily import arbitrary files from within a package to run quick tests, automatically grabbing dependencies as necessary
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Launch a Python REPL with in-tree modules already imported.
Missing third-party dependencies are resolved automatically: the target's import
graph is walked, anything unimportable is installed into a per-project cache venv
that is appended to sys.path, and an import/install loop mops up whatever the
static scan missed. Versions come from the project's uv.lock and packages from the
project's declared index, so the REPL matches what the code actually runs against.
Nothing is ever installed into the project's own venv.
Multiple targets, package dirs, and relative or absolute paths all work.
The shebang is a plain interpreter rather than ``uv run --script`` on purpose: the
REPL runs under the interpreter that owns the target code (see find_interpreter),
not under an ephemeral env for this script.
"""
from __future__ import annotations
import argparse
import ast
import hashlib
import json
import keyword
import os
import re
import shutil
import subprocess
import sys
from collections.abc import Sequence
from pathlib import Path
from typing import NamedTuple
# Import names whose PyPI distribution name differs, so installs and hints target
# the right package.
DIST_NAMES = {
"OpenSSL": "pyopenssl",
"PIL": "pillow",
"attr": "attrs",
"bs4": "beautifulsoup4",
"cv2": "opencv-python",
"dateutil": "python-dateutil",
"dns": "dnspython",
"google": "protobuf",
"jwt": "pyjwt",
"magic": "python-magic",
"pkg_resources": "setuptools",
"ruamel": "ruamel.yaml",
"snowflake": "snowflake-connector-python",
"sqlalchemy": "SQLAlchemy",
"yaml": "pyyaml",
"zoneinfo": "backports.zoneinfo",
}
# Files that mark a distribution/repo root. A directory holding one of these is
# never treated as an implicit namespace package to climb through.
ROOT_MARKERS = (
"setup.py",
"setup.cfg",
"pyproject.toml",
"tox.ini",
"requirements.txt",
"Makefile",
".git",
".hg",
".svn",
)
CACHE_HOME = (
Path(os.environ.get("XDG_CACHE_HOME", str(Path.home() / ".cache"))) / "pyrepl"
)
MAX_INSTALL_ROUNDS = 12
class Target(NamedTuple):
root: Path # directory to place on sys.path
dotted: str # full dotted module name
alias: str # name bound in the REPL
def is_importable_name(name: str) -> bool:
return name.isidentifier() and not keyword.iskeyword(name)
def has_root_marker(directory: Path) -> bool:
return any((directory / marker).exists() for marker in ROOT_MARKERS)
def is_package_component(directory: Path) -> bool:
"""Whether this directory should contribute a component to the dotted path.
A regular package (has ``__init__.py``) always does. A directory without one
may still be a PEP 420 namespace package, so climb through it as long as its
name is a usable identifier and it does not look like a project root.
"""
if (directory / "__init__.py").exists():
return True
return is_importable_name(directory.name) and not has_root_marker(directory)
def resolve(raw: str, forced_root: Path | None) -> Target:
path = Path(raw).expanduser().resolve()
if not path.exists():
raise ValueError("{}: no such file or directory".format(raw))
if forced_root is not None and forced_root not in path.parents:
raise ValueError(
"{}: --root {} is not an ancestor of this target".format(raw, forced_root)
)
if path.is_dir():
if not (path / "__init__.py").exists() and forced_root is None:
raise ValueError(
"{}: not a package (no __init__.py); pass --root to import it "
"as a namespace package".format(path)
)
parts = [path.name]
current = path.parent
elif path.suffix == ".py":
if path.name == "__init__.py":
# Point at the package itself rather than a bare "__init__" module.
parts = [path.parent.name]
current = path.parent.parent
else:
parts = [path.stem]
current = path.parent
else:
raise ValueError("{}: not a .py file or a package directory".format(raw))
stops = {Path(path.anchor), Path.home()}
while current not in stops and current != current.parent:
if forced_root is not None:
if current == forced_root:
break
elif not is_package_component(current):
break
if not is_importable_name(current.name):
raise ValueError(
"{}: directory name is not a valid Python identifier; pass "
"--root to set the import root explicitly".format(current)
)
parts.insert(0, current.name)
current = current.parent
if forced_root is not None and current != forced_root:
raise ValueError(
"{}: --root {} is not an ancestor of this target".format(raw, forced_root)
)
return Target(root=current, dotted=".".join(parts), alias=parts[-1])
def dedupe_aliases(targets: Sequence[Target]) -> list[Target]:
"""Disambiguate colliding aliases by prefixing the parent package."""
seen: dict[str, list[Target]] = {}
for target in targets:
seen.setdefault(target.alias, []).append(target)
out = []
for target in targets:
if len(seen[target.alias]) == 1:
out.append(target)
continue
parts = target.dotted.split(".")
alias = "_".join(parts[-2:]) if len(parts) > 1 else parts[-1]
out.append(target._replace(alias=alias))
return out
def find_interpreter(roots: Sequence[Path]) -> tuple[str, str]:
"""Pick the interpreter most likely to have the target's dependencies.
Returns (path, reason). An explicitly activated venv wins; otherwise look for
a project venv at or above the import root, since that is the environment the
code under test actually runs in. Falls back to the running interpreter.
"""
active = os.environ.get("VIRTUAL_ENV")
if active:
candidate = Path(active) / "bin" / "python"
if candidate.exists():
return str(candidate), "$VIRTUAL_ENV"
for root in roots:
for directory in [root] + list(root.parents):
if directory == Path(directory.anchor):
break
for name in (".venv", "venv"):
candidate = directory / name / "bin" / "python"
if candidate.exists():
return str(candidate), "found {}".format(candidate.parent.parent)
return sys.executable, "running interpreter"
def locate_module(dotted: str, roots: Sequence[Path]) -> Path | None:
"""Map a dotted name to an in-tree source file, or None if not in-tree."""
parts = dotted.split(".")
for root in roots:
base = root.joinpath(*parts)
package_init = base / "__init__.py"
if package_init.is_file():
return package_init
module = base.with_suffix(".py")
if module.is_file():
return module
return None
def _package_of(dotted: str, path: Path) -> str:
"""The package a relative import inside `path` is resolved against."""
if path.name == "__init__.py":
return dotted
return dotted.rpartition(".")[0]
def scan_imports(targets: Sequence[Target], roots: Sequence[Path]) -> set[str]:
"""Collect top-level external import names reachable from the targets.
Walks in-tree modules transitively -- following relative imports as well as
absolute ones that resolve inside the tree -- so a dependency pulled in by a
sibling module (``..schemaops`` needing sqlalchemy) is found without importing
anything.
"""
external: set[str] = set()
seen: set[str] = set()
queue = [t.dotted for t in targets]
while queue:
dotted = queue.pop()
if dotted in seen:
continue
seen.add(dotted)
path = locate_module(dotted, roots)
if path is None:
external.add(dotted.split(".")[0])
continue
try:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
except (OSError, SyntaxError, UnicodeDecodeError):
continue
package = _package_of(dotted, path)
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
queue.append(alias.name)
elif isinstance(node, ast.ImportFrom):
if node.level:
# Strip one package level per extra dot beyond the first.
base = package
for _ in range(node.level - 1):
base = base.rpartition(".")[0]
absolute = ".".join(p for p in (base, node.module) if p)
if absolute:
queue.append(absolute)
elif node.module:
queue.append(node.module)
# A package directory that is itself in-tree may appear via its parent name.
return {name for name in external if name}
def find_upwards(start: Path, filename: str) -> Path | None:
for directory in [start] + list(start.parents):
candidate = directory / filename
if candidate.is_file():
return candidate
if directory == Path(directory.anchor):
break
return None
def default_index_url(root: Path) -> tuple[str, Path] | None:
"""The project's default package index, if it declares one."""
pyproject = find_upwards(root, "pyproject.toml")
if pyproject is None:
return None
try:
text = pyproject.read_text(encoding="utf-8")
except OSError:
return None
for block in re.findall(r"\[\[tool\.uv\.index\]\](.*?)(?=\n\[|\Z)", text, re.S):
if re.search(r"^\s*default\s*=\s*true", block, re.M):
match = re.search(r'^\s*url\s*=\s*"([^"]+)"', block, re.M)
if match:
return match.group(1), pyproject
return None
def normalize_dist(name: str) -> str:
return re.sub(r"[-_.]+", "-", name).lower()
def locked_versions(root: Path) -> tuple[dict[str, set[str]], Path | None]:
"""Map normalized distribution name -> versions pinned in uv.lock."""
lock = find_upwards(root, "uv.lock")
if lock is None:
return {}, None
try:
text = lock.read_text(encoding="utf-8")
except OSError:
return {}, None
versions: dict[str, set[str]] = {}
pattern = re.compile(
r'\[\[package\]\]\s*\nname\s*=\s*"([^"]+)"\s*\nversion\s*=\s*"([^"]+)"'
)
for name, version in pattern.findall(text):
versions.setdefault(normalize_dist(name), set()).add(version)
return versions, lock
def dist_for(module: str) -> str:
return DIST_NAMES.get(module, module)
def pin(dist: str, versions: dict[str, set[str]]) -> tuple[str, str | None]:
"""Turn a distribution name into an install spec, pinned when unambiguous."""
found = versions.get(normalize_dist(dist))
if not found:
return dist, None
if len(found) > 1:
# Several versions coexist in the lock (different python markers); pinning
# to a guess would be worse than letting the resolver choose.
return dist, "multiple locked versions {}".format(sorted(found))
return "{}=={}".format(dist, sorted(found)[0]), None
def probe_missing(
interpreter: str, roots: Sequence[Path], extra: Sequence[Path], names: set[str]
) -> list[str]:
"""Which of `names` the interpreter cannot already import."""
if not names:
return []
script = (
"import importlib.util, json, sys\n"
"sys.path[:0] = json.loads(sys.argv[1])\n"
"sys.path.extend(json.loads(sys.argv[2]))\n"
"missing = []\n"
"for name in json.loads(sys.argv[3]):\n"
" try:\n"
" if importlib.util.find_spec(name) is None:\n"
" missing.append(name)\n"
" except Exception:\n"
" missing.append(name)\n"
"print(json.dumps(missing))\n"
)
result = subprocess.run(
[
interpreter,
"-c",
script,
json.dumps([str(r) for r in roots]),
json.dumps([str(p) for p in extra]),
json.dumps(sorted(names)),
],
capture_output=True,
text=True,
)
if result.returncode != 0:
return sorted(names)
try:
return json.loads(result.stdout.strip() or "[]")
except ValueError:
return sorted(names)
def probe_import(
interpreter: str, roots: Sequence[Path], extra: Sequence[Path], dotted: str
) -> str | None:
"""Actually import `dotted`; return the missing module name, if any."""
script = (
"import importlib, json, sys\n"
"sys.path[:0] = json.loads(sys.argv[1])\n"
"sys.path.extend(json.loads(sys.argv[2]))\n"
"try:\n"
" importlib.import_module(sys.argv[3])\n"
"except ModuleNotFoundError as exc:\n"
" print('MISSING:' + (exc.name or ''))\n"
"except Exception:\n"
" pass\n"
)
result = subprocess.run(
[
interpreter,
"-c",
script,
json.dumps([str(r) for r in roots]),
json.dumps([str(p) for p in extra]),
dotted,
],
capture_output=True,
text=True,
)
for line in result.stdout.splitlines():
if line.startswith("MISSING:"):
name = line[len("MISSING:") :].split(".")[0]
return name or None
return None
def interpreter_tag(interpreter: str) -> str:
result = subprocess.run(
[interpreter, "-c", "import sys; print('%d.%d' % sys.version_info[:2])"],
capture_output=True,
text=True,
)
return result.stdout.strip() or "unknown"
def cache_venv_path(root: Path, tag: str) -> Path:
digest = hashlib.sha256(str(root).encode("utf-8")).hexdigest()[:8]
return CACHE_HOME / "{}-{}-py{}".format(root.name or "root", digest, tag)
def ensure_cache_venv(interpreter: str, root: Path, tag: str, quiet: bool) -> Path:
"""A venv, matched to the base interpreter, holding only auto-pulled deps.
Deliberately separate from the project's own venv: this tool must never
mutate an environment the user maintains.
"""
path = cache_venv_path(root, tag)
python = path / "bin" / "python"
if python.exists():
return path
uv = shutil.which("uv")
if uv is None:
raise ValueError("automatic dependency install requires uv on PATH")
path.parent.mkdir(parents=True, exist_ok=True)
if not quiet:
print("pyrepl: creating dependency cache {}".format(path), file=sys.stderr)
result = subprocess.run(
[uv, "venv", "--python", interpreter, str(path)],
capture_output=True,
text=True,
)
if not python.exists():
raise ValueError(
"could not create cache venv: {}".format(
(result.stderr or result.stdout).strip().splitlines()[-1:] or "unknown"
)
)
return path
def site_packages(venv: Path) -> Path | None:
"""Layout is fixed for a venv, so glob it rather than spawning an interpreter."""
for candidate in sorted(venv.glob("lib/python*/site-packages")):
return candidate
return None
def uv_install(
venv: Path, specs: Sequence[str], index: str | None, cwd: Path, quiet: bool
) -> tuple[bool, str]:
uv = shutil.which("uv")
if uv is None:
return False, "uv not on PATH"
# Explicit --python keeps uv pointed at the cache venv; it never discovers or
# syncs the surrounding project the way a bare `uv run` would.
argv = [uv, "pip", "install", "--python", str(venv / "bin" / "python")]
if index:
argv += ["--default-index", index]
argv += list(specs)
if not quiet:
print("pyrepl: installing {}".format(" ".join(specs)), file=sys.stderr)
result = subprocess.run(argv, capture_output=True, text=True, cwd=str(cwd))
if result.returncode != 0:
detail = (result.stderr or result.stdout).strip().splitlines()
return False, detail[-1] if detail else "install failed"
return True, ""
def autoresolve(
targets: Sequence[Target],
roots: Sequence[Path],
interpreter: str,
with_deps: Sequence[str],
quiet: bool,
) -> tuple[list[Path], list[str]]:
"""Install whatever the targets need, returning (extra sys.path, notes)."""
notes: list[str] = []
root = roots[0]
versions, lock = locked_versions(root)
index_info = default_index_url(root)
index = index_info[0] if index_info else None
announced = [False]
def announce() -> None:
if quiet or announced[0]:
return
announced[0] = True
if lock:
print("pyrepl: pinning versions from {}".format(lock), file=sys.stderr)
print("pyrepl: index {}".format(index or "default (PyPI)"), file=sys.stderr)
extra: list[Path] = []
venv: Path | None = None
tag = interpreter_tag(interpreter)
installed: set[str] = set()
def ensure() -> Path:
nonlocal venv, extra
if venv is None:
venv = ensure_cache_venv(interpreter, root, tag, quiet)
packages = site_packages(venv)
if packages:
extra = [packages]
return venv
def install(specs: Sequence[str]) -> tuple[bool, str]:
announce()
return uv_install(ensure(), specs, index, root, quiet)
existing = cache_venv_path(root, tag)
if (existing / "bin" / "python").exists():
ensure()
if with_deps:
specs = []
for dep in with_deps:
spec, note = pin(dep, versions)
specs.append(spec)
if note:
notes.append("{}: {}".format(dep, note))
ok, err = install(specs)
if not ok:
notes.append("install failed: {}".format(err))
pending = [t.dotted for t in targets]
missing_first = [probe_import(interpreter, roots, extra, d) for d in pending]
if not any(missing_first):
return extra, notes
scanned = scan_imports(targets, roots)
absent = probe_missing(interpreter, roots, extra, scanned)
if absent:
specs = []
for module in absent:
spec, note = pin(dist_for(module), versions)
specs.append(spec)
if note:
notes.append("{}: {}".format(module, note))
installed.update(absent)
ok, err = install(specs)
if not ok:
notes.append("batch install failed ({}); retrying individually".format(err))
for module, spec in zip(absent, specs):
ok, err = install([spec])
if not ok:
notes.append("could not install {}: {}".format(spec, err))
for _ in range(MAX_INSTALL_ROUNDS):
stuck = None
for target in targets:
module = probe_import(interpreter, roots, extra, target.dotted)
if module and module not in installed:
stuck = module
break
if stuck is None:
break
installed.add(stuck)
spec, note = pin(dist_for(stuck), versions)
if note:
notes.append("{}: {}".format(stuck, note))
ok, err = install([spec])
if not ok:
notes.append("could not install {} (for {}): {}".format(spec, stuck, err))
break
return extra, notes
def build_bootstrap(
targets: Sequence[Target],
extra: Sequence[Path],
star: bool,
banner: bool,
) -> str:
"""Python source run via ``-c``, whose globals become the REPL namespace."""
roots: list[str] = []
for target in targets:
root = str(target.root)
if root not in roots:
roots.append(root)
pairs = [(t.alias, t.dotted) for t in targets]
return """\
import importlib, sys
_dist_names = {dist_names!r}
_roots = {roots!r}
_extra = {extra!r}
_pairs = {pairs!r}
_star = {star!r}
_banner = {banner!r}
for _root in reversed(_roots):
if _root not in sys.path:
sys.path.insert(0, _root)
# Appended, not inserted: the project's own environment must win over anything
# pyrepl auto-installed.
for _dir in _extra:
if _dir not in sys.path:
sys.path.append(_dir)
_loaded = {{}}
_failed = []
for _alias, _dotted in _pairs:
try:
_loaded[_alias] = importlib.import_module(_dotted)
except Exception as _exc:
_missing = (
getattr(_exc, "name", None) if isinstance(_exc, ImportError) else None
)
if _missing and _missing.split(".")[0] in _dotted.split("."):
_missing = None # in-tree import error, not a dependency gap
_failed.append(
(_dotted, "{{}}: {{}}".format(type(_exc).__name__, _exc), _missing)
)
globals().update(_loaded)
if _star:
for _mod in _loaded.values():
_names = getattr(_mod, "__all__", None)
if _names is None:
_names = [_n for _n in vars(_mod) if not _n.startswith("_")]
globals().update({{_n: getattr(_mod, _n) for _n in _names}})
def reload_all():
\"\"\"Re-import every target module after editing it on disk.\"\"\"
for _alias, _mod in list(_loaded.items()):
_loaded[_alias] = importlib.reload(_mod)
globals()[_alias] = _loaded[_alias]
return sorted(_loaded)
if _banner:
_w = max([len(_a) for _a, _ in _pairs] + [0])
for _alias, _dotted in _pairs:
_mark = " " if _alias in _loaded else "! "
print("{{}}{{}} {{}}".format(_mark, _alias.ljust(_w), _dotted), file=sys.stderr)
print("reload_all() to re-import after editing", file=sys.stderr)
for _dotted, _err, _missing in _failed:
print(
"pyrepl: failed to import {{}}\\n {{}}".format(_dotted, _err),
file=sys.stderr,
)
if _missing:
print(
" install it with: pyrepl --with {{}} ...".format(
_dist_names.get(_missing, _missing)
),
file=sys.stderr,
)
for _name in ("_root", "_dir", "_alias", "_dotted", "_mod", "_names", "_w", "_mark"):
globals().pop(_name, None)
del _name
""".format(
roots=roots,
extra=[str(p) for p in extra],
pairs=pairs,
star=star,
banner=banner,
dist_names=DIST_NAMES,
)
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(
prog="pyrepl",
description="Open a REPL with in-tree modules imported, pulling any "
"missing third-party dependencies automatically.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="example:\n pyrepl path/to/pkg/mod.py another/pkg\n",
)
parser.add_argument(
"targets",
nargs="+",
metavar="PATH",
help="one or more .py files or package directories (relative or absolute)",
)
parser.add_argument(
"--root",
metavar="DIR",
help="force the sys.path import root instead of inferring it",
)
parser.add_argument(
"--python",
metavar="PATH",
help="interpreter to launch (default: the venv owning the target code)",
)
parser.add_argument(
"--with",
dest="with_deps",
action="append",
default=[],
metavar="PKG",
help="extra distribution to install up front (repeatable)",
)
parser.add_argument(
"--no-deps",
dest="auto_deps",
action="store_false",
help="do not install anything; just report what is missing",
)
parser.add_argument(
"--clear-cache",
action="store_true",
help="delete this project's dependency cache venv and rebuild it",
)
parser.add_argument(
"-s",
"--star",
action="store_true",
help="also splat each module's public names into the REPL namespace",
)
parser.add_argument(
"-q",
"--quiet",
action="store_true",
help="suppress the startup banner and install progress",
)
parser.add_argument(
"-n",
"--dry-run",
action="store_true",
help="report resolution and the dependency plan; install and launch nothing",
)
args = parser.parse_args(argv)
forced_root = None
if args.root:
forced_root = Path(args.root).expanduser().resolve()
if not forced_root.is_dir():
print(
"pyrepl: --root {}: not a directory".format(args.root), file=sys.stderr
)
return 2
targets = []
for raw in args.targets:
try:
targets.append(resolve(raw, forced_root))
except ValueError as exc:
print("pyrepl: {}".format(exc), file=sys.stderr)
return 2
targets = dedupe_aliases(targets)
roots: list[Path] = []
for target in targets:
if target.root not in roots:
roots.append(target.root)
if args.python:
interpreter, reason = args.python, "--python"
else:
interpreter, reason = find_interpreter(roots)
interpreter = shutil.which(interpreter) or interpreter
if not os.path.exists(interpreter):
print("pyrepl: {}: interpreter not found".format(interpreter), file=sys.stderr)
return 2
if args.clear_cache:
stale = cache_venv_path(roots[0], interpreter_tag(interpreter))
if stale.is_dir():
shutil.rmtree(stale)
print("pyrepl: removed {}".format(stale), file=sys.stderr)
if args.dry_run:
for target in targets:
print(
"{}\n root: {}\n as: {}".format(
target.dotted, target.root, target.alias
)
)
print("\ninterpreter: {} ({})".format(interpreter, reason))
scanned = scan_imports(targets, roots)
absent = probe_missing(interpreter, roots, [], scanned)
versions, lock = locked_versions(roots[0])
index = default_index_url(roots[0])
print("external imports found: {}".format(" ".join(sorted(scanned)) or "none"))
print("already importable: {}".format(len(scanned) - len(absent)))
if absent:
specs = [pin(dist_for(m), versions)[0] for m in absent]
print("would install: {}".format(" ".join(specs)))
print("lock: {}".format(lock or "none found"))
print("index: {}".format(index[0] if index else "default (PyPI)"))
print(
"cache: {}".format(cache_venv_path(roots[0], interpreter_tag(interpreter)))
)
return 0
extra: list[Path] = []
if args.auto_deps:
try:
extra, notes = autoresolve(
targets, roots, interpreter, args.with_deps, args.quiet
)
except ValueError as exc:
print("pyrepl: {}".format(exc), file=sys.stderr)
return 2
if not args.quiet:
for note in notes:
print("pyrepl: {}".format(note), file=sys.stderr)
if not args.quiet:
print("python: {} ({})".format(interpreter, reason), file=sys.stderr)
bootstrap = build_bootstrap(targets, extra, star=args.star, banner=not args.quiet)
env = dict(os.environ)
existing = env.get("PYTHONPATH", "")
root_strs = [str(r) for r in roots]
env["PYTHONPATH"] = os.pathsep.join(root_strs + ([existing] if existing else []))
os.execve(interpreter, [interpreter, "-i", "-c", bootstrap], env)
return 1
if __name__ == "__main__":
sys.exit(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment