https://kedro.org/ logo
#advanced-need-help
Title
# advanced-need-help
t

Tsakagur

05/09/2022, 3:30 PM
Copy code
python
def main(
    pipeline_name: str = typer.Option(default=None),
    tag: str = typer.Option(default=None),
    env: str = typer.Option(default="pipelines-aml"),
    scope: Optional[Scope] = typer.Option(default=None, help= "Must be present if one of the following CLI option is."),
    train_start_date: datetime = typer.Option(default=None, formats=[DATE_FORMAT]),
    train_end_date: datetime = typer.Option(default=None, formats=[DATE_FORMAT]),
    val_start_date: datetime = typer.Option(default=None, formats=[DATE_FORMAT]),
    val_end_date: datetime = typer.Option(default=None, formats=[DATE_FORMAT]),
) -> None:
    """
    Pico model training.
    Emulates ``kedro run`` command.

    """
    # Kedro session arguments
    session_kwargs = {"env": env}
    if scope and (train_start_date or train_end_date or val_start_date or val_end_date):
        extra_params = {}
        if train_start_date:
            extra_params[f"{scope}_train_start_date"] = train_start_date
        if train_end_date:
            extra_params[f"{scope}_train_end_date"] = train_end_date
        if val_start_date:
            extra_params[f"{scope}_val_start_date"] = val_start_date
        if val_end_date:
            extra_params[f"{scope}_val_end_date"] = val_end_date
        session_kwargs["extra_params"] = extra_params
    print(f'{session_kwargs=}')

    # Kedro run arguments
    run_kwargs = {}
    if pipeline_name:
        run_kwargs["pipeline_name"] = pipeline_name
    if tag:
        run_kwargs["tags"] = [tag]
    print(f'{run_kwargs=}')

    metadata = bootstrap_project(Path.cwd())
    with KedroSession.create(metadata.package_name, project_path=Path.cwd(), **session_kwargs) as session:
        session.run(**run_kwargs)
2 Views