mirror of
https://github.com/daylinmorgan/viv.git
synced 2024-11-12 12:13:15 -06:00
Compare commits
7 commits
2dd4b35ee7
...
56b9698467
Author | SHA1 | Date | |
---|---|---|---|
56b9698467 | |||
24d19dd4d6 | |||
16c538c6d7 | |||
56013e9d87 | |||
9229337f29 | |||
d1ea8c3868 | |||
0996628a97 |
3 changed files with 168 additions and 55 deletions
|
@ -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
110
examples/dictionary.py
Normal 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()
|
111
src/viv/viv.py
111
src/viv/viv.py
|
@ -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,
|
||||
|
|
Loading…
Reference in a new issue