def install()

in metaflow/cmd/develop/stubs.py [0:0]
78 lines of code
16 McCabe index (conditional complexity)

def install(ctx: Any, force: bool):
    """
    Generates the Python stubs for Metaflow considering the installed version of
    Metaflow. The stubs will be generated if they do not exist or do not match the
    current version of Metaflow and installed in the Python environment.
    """
    try:
        import build
    except ImportError:
        raise RuntimeError(
            "Installing stubs requires 'build' -- please install it and try again"
        )

    dist_packages, paths = get_packages_for_stubs()
    if paths:
        raise RuntimeError(
            "Cannot install stubs when metaflow-stubs is already provided by a directory. "
            "Please remove the following and try again: %s" % ", ".join(paths)
        )

    if len(dist_packages) == 1:
        if internal_check(dist_packages[0][1])[1] == True and not force:
            if ctx.obj.quiet:
                ctx.obj.echo_always("already_installed")
            else:
                ctx.obj.echo(
                    "Metaflow stubs are already installed and valid -- use --force to reinstall"
                )
            return
    mf_version, _ = get_mf_version(True)
    with tempfile.TemporaryDirectory() as tmp_dir:
        with open(os.path.join(tmp_dir, "setup.py"), "w") as f:
            f.write(
                f"""
from setuptools import setup, find_namespace_packages
setup(
    include_package_data=True,
    name="metaflow-stubs",
    version="{mf_version}",
    description="Metaflow: More Data Science, Less Engineering",
    author="Metaflow Developers",
    author_email="help@metaflow.org",
    license="Apache Software License",
    packages=find_namespace_packages(),
    package_data={{"metaflow-stubs": ["generated_for.txt", "py.typed", "**/*.pyi"]}},
    install_requires=["metaflow=={mf_version}"],
    python_requires=">=3.5.2",
)
                """
            )
        with open(os.path.join(tmp_dir, "MANIFEST.in"), "w") as f:
            f.write(
                """
include metaflow-stubs/generated_for.txt
include metaflow-stubs/py.typed
global-include *.pyi
                """
            )

        StubGenerator(os.path.join(tmp_dir, "metaflow-stubs")).write_out()

        subprocess.check_call(
            [sys.executable, "-m", "build", "--wheel"],
            cwd=tmp_dir,
            stderr=subprocess.DEVNULL if ctx.obj.quiet else None,
            stdout=subprocess.DEVNULL if ctx.obj.quiet else None,
        )

        if dist_packages:
            # We need to uninstall all the other packages first
            pkgs_to_remove = [p[0] for p in dist_packages]
            ctx.obj.echo(
                "Uninstalling existing packages providing metaflow-stubs: %s"
                % ", ".join(pkgs_to_remove)
            )

            subprocess.check_call(
                [
                    sys.executable,
                    "-m",
                    "pip",
                    "uninstall",
                    "-y",
                    *pkgs_to_remove,
                ],
                cwd=tmp_dir,
                stderr=subprocess.DEVNULL if ctx.obj.quiet else None,
                stdout=subprocess.DEVNULL if ctx.obj.quiet else None,
            )

        subprocess.check_call(
            [
                sys.executable,
                "-m",
                "pip",
                "install",
                "--force-reinstall",
                "--no-deps",
                "--no-index",
                "--find-links",
                os.path.join(tmp_dir, "dist"),
                "metaflow-stubs",
            ],
            cwd=tmp_dir,
            stderr=subprocess.DEVNULL if ctx.obj.quiet else None,
            stdout=subprocess.DEVNULL if ctx.obj.quiet else None,
        )
    if ctx.obj.quiet:
        ctx.obj.echo_always("installed")
    else:
        ctx.obj.echo("Metaflow stubs successfully installed")