4764305061

Build Type

github

Committed by GitHub

Commit Message

Merge 68c2654fb into 556449cd0

Pull Request Pull Request #25: Update dask4dvc arguments

Run Details

132 of 132 new or added lines in 5 files covered. (100.0%)

237 of 290 relevant lines covered (81.72%)

0.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.95

/dask4dvc/cli/main.py

"""All methods that come directly from 'dask4dvc' CLI interace."""

import importlib.metadata
import logging

import dask.distributed
import typer
import typing
from dask4dvc import methods, utils


app = typer.Typer()

log = logging.getLogger(__name__)


class Help:
    """Collect typer help strings that are used multiple times."""

    address: str = (
        "This can be the address of a DASK Scheduler server like '127.0.0.1:31415'. If"
        " 'None' Dask will launch a new Server."
    )
    parallel: str = (
        "Split the DVC Graph into individual Nodes and run them in parallel if possible."
    )
    leave: str = "Ask before stopping the client"
    config: str = "path to config file, e.g. 'dask4dvc.yaml'"
    retries: str = "Number of retries to acquire dvc lock."
    max_workers: str = (
        "Maximum number of workers to use. Using '1' will be the same as 'dvc repro' but"
        " slower."
    )


@app.command()
def repro(
    targets: typing.List[str] = typer.Argument(None),
    address: str = typer.Option(None, help=Help.address),
    leave: bool = typer.Option(True, help=Help.leave),
    config: str = typer.Option(None, help=Help.config),
    max_workers: int = typer.Option(None, help=Help.max_workers),
    retries: int = typer.Option(10, help=Help.retries),
    force: bool = typer.Option(False, "--force/", "-f/", help="use `dvc repro --force`"),
) -> None:
    """Replicate 'dvc repro' command using dask."""
    utils.CONFIG.retries = retries

    if config is not None:
        assert address is None, "Can not use address and config file"
        address = utils.dask.get_cluster_from_config(config)

    with dask.distributed.Client(address) as client:
        if max_workers is not None:
            client.cluster.adapt(minimum=1, maximum=max_workers)
        log.info(client)
        results = methods.parallel_submit(client, targets=targets, force=force)

        utils.dask.wait_for_futures(results)
        if not leave:
            utils.main.wait()


@app.command()
def run() -> None:
    """Run DVC experiments in parallel using dask."""
    typer.echo(
        "'dask4dvc run' was removed due to instability. Latest version including this"
        " feature is 'pip install dask4dvc==0.1.4'."
    )
    raise typer.Exit()


def version_callback(value: bool) -> None:
    """Get the installed dask4dvc version."""
    if value:
        typer.echo(f"dask4dvc {importlib.metadata.version('dask4dvc')}")
        raise typer.Exit()


@app.callback()
def main(
    version: bool = typer.Option(
        None, "--version", callback=version_callback, is_eager=True
    ),
) -> None:
    """Dask4DVC CLI callback.

    Run the DVC graph or DVC experiments in parallel using dask.

    """
    _ = version  # this would be greyed out otherwise

1	"""All methods that come directly from 'dask4dvc' CLI interace."""	1✔
2
3	import importlib.metadata	1✔
4	import logging	1✔
5
6	import dask.distributed	1✔
7	import typer	1✔
8	import typing	1✔
9	from dask4dvc import methods, utils	1✔
10
11
12	app = typer.Typer()	1✔
13
14	log = logging.getLogger(__name__)	1✔
15
16
17	class Help:	1✔
18	"""Collect typer help strings that are used multiple times."""	1✔
19
20	address: str = (	1✔
21	"This can be the address of a DASK Scheduler server like '127.0.0.1:31415'. If"
22	" 'None' Dask will launch a new Server."
23	)
24	parallel: str = (	1✔
25	"Split the DVC Graph into individual Nodes and run them in parallel if possible."
26	)
27	leave: str = "Ask before stopping the client"	1✔
28	config: str = "path to config file, e.g. 'dask4dvc.yaml'"	1✔
29	retries: str = "Number of retries to acquire dvc lock."	1✔
30	max_workers: str = (	1✔
31	"Maximum number of workers to use. Using '1' will be the same as 'dvc repro' but"
32	" slower."
33	)
34
35
36	@app.command()	1✔
37	def repro(	1✔
38	targets: typing.List[str] = typer.Argument(None),
39	address: str = typer.Option(None, help=Help.address),
40	leave: bool = typer.Option(True, help=Help.leave),
41	config: str = typer.Option(None, help=Help.config),
42	max_workers: int = typer.Option(None, help=Help.max_workers),
43	retries: int = typer.Option(10, help=Help.retries),
44	force: bool = typer.Option(False, "--force/", "-f/", help="use `dvc repro --force`"),
45	) -> None:
46	"""Replicate 'dvc repro' command using dask."""
47	utils.CONFIG.retries = retries	1✔
48
49	if config is not None:	1✔
50	assert address is None, "Can not use address and config file"	×
51	address = utils.dask.get_cluster_from_config(config)	×
52
53	with dask.distributed.Client(address) as client:	1✔
54	if max_workers is not None:	1✔
55	client.cluster.adapt(minimum=1, maximum=max_workers)	×
56	log.info(client)	1✔
57	results = methods.parallel_submit(client, targets=targets, force=force)	1✔
58
59	utils.dask.wait_for_futures(results)	1✔
60	if not leave:	1✔
61	utils.main.wait()	×
62
63
64	@app.command()	1✔
65	def run() -> None:	1✔
66	"""Run DVC experiments in parallel using dask."""
67	typer.echo(	×
68	"'dask4dvc run' was removed due to instability. Latest version including this"
69	" feature is 'pip install dask4dvc==0.1.4'."
70	)
71	raise typer.Exit()	×
72
73
74	def version_callback(value: bool) -> None:	1✔
75	"""Get the installed dask4dvc version."""
76	if value:	1✔
77	typer.echo(f"dask4dvc {importlib.metadata.version('dask4dvc')}")	×
78	raise typer.Exit()	×
79
80
81	@app.callback()	1✔
82	def main(	1✔
83	version: bool = typer.Option(
84	None, "--version", callback=version_callback, is_eager=True
85	),
86	) -> None:
87	"""Dask4DVC CLI callback.
88
89	Run the DVC graph or DVC experiments in parallel using dask.
90
91	"""
92	_ = version # this would be greyed out otherwise	1✔

zincware / dask4dvc / 4764305061

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous