2023-05-08 03:55:02 +00:00
|
|
|
import importlib
|
|
|
|
import typer
|
|
|
|
import sys
|
2023-05-08 04:07:47 +00:00
|
|
|
from typing import List, Optional
|
2023-05-08 03:55:02 +00:00
|
|
|
from typing_extensions import Annotated
|
|
|
|
|
|
|
|
app = typer.Typer()
|
|
|
|
|
|
|
|
|
|
|
|
def _load_recipe(dotted_path: str):
|
|
|
|
sys.path.append(".")
|
|
|
|
path, name = dotted_path.rsplit(".", 1)
|
|
|
|
mod = importlib.import_module(path)
|
|
|
|
return getattr(mod, name)
|
|
|
|
|
|
|
|
|
|
|
|
@app.command()
|
|
|
|
def reset(recipe: Annotated[str, typer.Option(...)]):
|
|
|
|
mod = _load_recipe(recipe)
|
|
|
|
mod.reset()
|
|
|
|
|
|
|
|
|
|
|
|
@app.command()
|
|
|
|
def show(recipe: Annotated[str, typer.Option(...)]):
|
|
|
|
mod = _load_recipe(recipe)
|
|
|
|
mod.show()
|
|
|
|
|
|
|
|
|
|
|
|
@app.command()
|
2023-05-08 04:07:47 +00:00
|
|
|
def run(
|
|
|
|
recipe: Annotated[str, typer.Option(...)],
|
|
|
|
input: Annotated[Optional[List[str]], typer.Option(...)] = None,
|
|
|
|
):
|
2023-05-08 03:55:02 +00:00
|
|
|
mod = _load_recipe(recipe)
|
2023-05-08 04:07:47 +00:00
|
|
|
for input_str in input:
|
|
|
|
beaker, filename = input_str.split("=")
|
|
|
|
mod.csv_to_beaker(filename, beaker)
|
2023-05-08 03:55:02 +00:00
|
|
|
mod.run_once()
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
app()
|