Compare commits

...

7 commits

3 changed files with 168 additions and 55 deletions

View file

@ -18,7 +18,7 @@ def _viv_use(*pkgs, track_exe=False, name=""):
_id = sha256.hexdigest()
if (env := cache / (name if name else _id)) not in cache.glob("*/") or force:
sys.stderr.write(f"generating new vivenv -> {env.name}\n")
venv.EnvBuilder(with_pip=True, clear=True).create(env)
venv.create(env, symlinks=True, with_pip=True, clear=True).create(env)
(env / "pip.conf").write_text("[global]\ndisable-pip-version-check=true")
run_kw = dict(zip(("stdout", "stderr"), ((None,) * 2 if verbose else (-1, 2))))
p = run([env / "bin" / "pip", "install", "--force-reinstall", *spec], **run_kw)

110
examples/dictionary.py Normal file
View file

@ -0,0 +1,110 @@
#!/usr/bin/env python3
"""
Adapted from https://github.com/Textualize/textual/blob/main/examples/dictionary.py
"""
from __future__ import annotations
__import__("viv").use("textual", "httpx") # noqa
try:
import httpx
except ImportError:
raise ImportError("Please install httpx with 'pip install httpx' ")
from textual import work
from textual.app import App, ComposeResult
from textual.containers import VerticalScroll
from textual.widgets import Input, Markdown
class DictionaryApp(App):
"""Searches ab dictionary API as-you-type."""
# CSS_PATH = "dictionary.css"
DEFAULT_CSS = """
Screen {
background: $panel;
}
Input {
dock: top;
margin: 1 0;
}
#results {
width: 100%;
height: auto;
}
#results-container {
background: $background 50%;
margin: 0 0 1 0;
height: 100%;
overflow: hidden auto;
border: tall $background;
}
#results-container:focus {
border: tall $accent;
}
"""
def compose(self) -> ComposeResult:
yield Input(placeholder="Search for a word")
with VerticalScroll(id="results-container"):
yield Markdown(id="results")
def on_mount(self) -> None:
"""Called when app starts."""
# Give the input focus, so we can start typing straight away
self.query_one(Input).focus()
async def on_input_changed(self, message: Input.Changed) -> None:
"""A coroutine to handle a text changed message."""
if message.value:
self.lookup_word(message.value)
else:
# Clear the results
self.query_one("#results", Markdown).update("")
@work(exclusive=True)
async def lookup_word(self, word: str) -> None:
"""Looks up a word."""
url = f"https://api.dictionaryapi.dev/api/v2/entries/en/{word}"
async with httpx.AsyncClient() as client:
response = await client.get(url)
try:
results = response.json()
except Exception:
self.query_one("#results", Markdown).update(response.text)
if word == self.query_one(Input).value:
markdown = self.make_word_markdown(results)
self.query_one("#results", Markdown).update(markdown)
def make_word_markdown(self, results: object) -> str:
"""Convert the results in to markdown."""
lines = []
if isinstance(results, dict):
lines.append(f"# {results['title']}")
lines.append(results["message"])
elif isinstance(results, list):
for result in results:
lines.append(f"# {result['word']}")
lines.append("")
for meaning in result.get("meanings", []):
lines.append(f"_{meaning['partOfSpeech']}_")
lines.append("")
for definition in meaning.get("definitions", []):
lines.append(f" - {definition['definition']}")
lines.append("---")
return "\n".join(lines)
if __name__ == "__main__":
app = DictionaryApp()
app.run()

View file

@ -43,6 +43,7 @@ from typing import (
List,
NoReturn,
Optional,
Sequence,
TextIO,
Tuple,
Type,
@ -50,7 +51,7 @@ from typing import (
from urllib.error import HTTPError
from urllib.request import urlopen
__version__ = "23.5a4-30-g3cf2dce-dev"
__version__ = "23.5a4-37-gf8ebd9d-dev"
class Spinner:
@ -171,7 +172,7 @@ class Ansi:
)
)
def subprocess(self, output: str) -> None:
def subprocess(self, command: List[str], output: str) -> None:
"""generate output for subprocess error
Args:
@ -179,9 +180,12 @@ class Ansi:
"""
if not output:
return
echo("subprocess output:")
error("subprocess failed")
echo("see below for command output", style="red")
echo(f"cmd:\n {' '.join(command)}", style="red")
new_output = [f"{self.red}->{self.end} {line}" for line in output.splitlines()]
sys.stdout.write("\n".join(new_output) + "\n")
echo("subprocess output:" + "\n".join(("", *new_output, "")), style="red")
def viv_preamble(self, style: str = "magenta", sep: str = "::") -> str:
return f"{self.cyan}viv{self.end}{self.__dict__[style]}{sep}{self.end}"
@ -218,7 +222,7 @@ to create/activate a vivenv:
_id = sha256.hexdigest()
if (env := cache / (name if name else _id)) not in cache.glob("*/") or force:
sys.stderr.write(f"generating new vivenv -> {env.name}\n")
venv.EnvBuilder(with_pip=True, clear=True).create(env)
venv.create(env, symlinks=True, with_pip=True, clear=True).create(env)
(env / "pip.conf").write_text("[global]\ndisable-pip-version-check=true")
run_kw = dict(zip(("stdout", "stderr"), ((None,) * 2 if verbose else (-1, 2))))
p = run([env / "bin" / "pip", "install", "--force-reinstall", *spec], **run_kw)
@ -561,9 +565,7 @@ def run(
)
if p.returncode != 0 and not ignore_error:
error("subprocess failed")
echo("see below for command output", style="red")
a.subprocess(p.stdout)
a.subprocess(command, p.stdout)
if clean_up_path and clean_up_path.is_dir():
shutil.rmtree(str(clean_up_path))
@ -688,12 +690,12 @@ class ViVenv:
if not quiet:
echo(f"new unique vivenv -> {self.name}")
with Spinner("creating vivenv"):
builder = venv.EnvBuilder(with_pip=True, clear=True)
builder.create(self.path)
venv.create(self.path, with_pip=True, clear=True, symlinks=True)
# add config to ignore pip version
with (self.path / "pip.conf").open("w") as f:
f.write("[global]\ndisable-pip-version-check = true")
(self.path / "pip.conf").write_text(
"[global]\ndisable-pip-version-check = true"
)
self.meta.created = str(datetime.today())
@ -808,24 +810,23 @@ def combined_spec(reqs: List[str], requirements: Path) -> List[str]:
def resolve_deps(args: Namespace) -> List[str]:
spec = combined_spec(args.reqs, args.requirements)
with tempfile.TemporaryDirectory(prefix="viv-") as tmpdir:
echo("generating frozen spec")
vivenv = ViVenv(spec, track_exe=False, path=Path(tmpdir))
cmd = [
"pip",
"install",
"--dry-run",
"--quiet",
"--ignore-installed",
"--report",
"-",
] + spec
vivenv.create(quiet=True)
# populate the environment for now use
# custom cmd since using requirements file
cmd = [
str(vivenv.path / "bin" / "pip"),
"install",
"--force-reinstall",
] + spec
report = json.loads(run(cmd, check_output=True, spinmsg="resolving depedencies"))
resolved_spec = [
f"{pkg['metadata']['name']}=={pkg['metadata']['version']}"
for pkg in report["install"]
]
run(cmd, spinmsg="resolving dependencies", clean_up_path=vivenv.path)
cmd = [str(vivenv.path / "bin" / "pip"), "freeze"]
resolved_spec = run(cmd, check_output=True)
return resolved_spec.splitlines()
return resolved_spec
def fetch_source(reference: str) -> str:
@ -1233,7 +1234,7 @@ class Arg:
class Cli:
args = {
"list": [
("list",): [
Arg(
"-f",
"--full",
@ -1247,7 +1248,7 @@ class Cli:
action="store_true",
),
],
"shim": [
("shim",): [
Arg(
"-f",
"--freeze",
@ -1262,8 +1263,8 @@ class Cli:
metavar="<path>",
),
],
"remove": [Arg("vivenv", help="name/hash of vivenv", nargs="*")],
("exe|pip", "exe|python"): [Arg("vivenv", help="name/hash of vivenv")],
("remove",): [Arg("vivenv", help="name/hash of vivenv", nargs="*")],
("exe|pip", "exe|python", "info"): [Arg("vivenv", help="name/hash of vivenv")],
("list", "info"): [
Arg(
"--json",
@ -1343,29 +1344,32 @@ class Cli:
)
],
}
cmds = (
"list",
(
"exe",
dict(
(
cmds := dict.fromkeys(
(
"list",
"shim",
"run",
"exe",
"remove",
"freeze",
"info",
"manage",
)
)
).update(
{
"exe": dict(
pip=dict(help="run cmd with pip"),
python=dict(help="run cmd with python"),
),
),
"remove",
"freeze",
"info",
(
"manage",
dict(
"manage": dict(
show=dict(help="show current installation", aliases=["s"]),
install=dict(help="install fresh viv", aliases=["i"]),
update=dict(help="update viv version", aliases=["u"]),
purge=dict(help="remove traces of viv", aliases=["p"]),
),
),
"shim",
"run",
}
)
def __init__(self, viv: Viv) -> None:
@ -1376,7 +1380,7 @@ class Cli:
self._add_args()
def _cmd_arg_group_map(self) -> None:
self.cmd_arg_group_map = {}
self.cmd_arg_group_map: Dict[str, List[Sequence[str] | str]] = {}
for grp in self.args:
if isinstance(grp, str):
self.cmd_arg_group_map.setdefault(grp, []).append(grp)
@ -1390,7 +1394,7 @@ class Cli:
def _add_args(self) -> None:
for grp, args in self.args.items():
for arg in args:
self.parsers.get(grp).add_argument(*arg.args, **arg.kwargs)
self.parsers[grp].add_argument(*arg.args, **arg.kwargs)
def _validate_args(self, args: Namespace) -> None:
if args.func.__name__ in ("freeze", "shim", "run"):
@ -1460,7 +1464,7 @@ class Cli:
return parser
def run(self):
def run(self) -> None:
self.parser.add_argument(
"-V",
"--version",
@ -1472,9 +1476,8 @@ class Cli:
metavar="<sub-cmd>", title="subcommands", required=True
)
for cmd in self.cmds:
if isinstance(cmd, tuple):
cmd, subcmds = cmd
for cmd, subcmds in self.cmds.items():
if subcmds:
subcmd_p = self._get_subcmd_parser(cmd_p, cmd)
subcmd_cmd_p = subcmd_p.add_subparsers(
title="subcommand", metavar="<sub-cmd>", required=True
@ -1483,7 +1486,7 @@ class Cli:
subcmd_cmd_p.add_parser(
subcmd,
parents=[
self.parsers.get(k)
self.parsers[k]
for k in self.cmd_arg_group_map[f"{cmd}|{subcmd}"]
],
**kwargs,