kedro/framework/cli/pipeline.py
def pull_package(package_path, env, alias):
    """Pull a modular pipeline package, unpack it and install the files to corresponding
    locations.
    """
    # pylint: disable=import-outside-toplevel
    import fsspec
    from kedro.io.core import get_protocol_and_path

    protocol, _ = get_protocol_and_path(package_path)
    filesystem = fsspec.filesystem(protocol)

    with tempfile.TemporaryDirectory() as temp_dir:
        temp_dir_path = Path(temp_dir).resolve()
        if package_path.endswith(".whl") and filesystem.exists(package_path):
            with filesystem.open(package_path) as fs_file:
                ZipFile(fs_file).extractall(temp_dir_path)
        else:
            python_call(
                "pip",
                ["download", "--no-deps", "--dest", str(temp_dir_path), package_path],
            )
            wheel_file = list(temp_dir_path.glob("*.whl"))
            # `--no-deps` should fetch only one wheel file, and CLI should fail if that's
            # not the case.
            if len(wheel_file) != 1:
                file_names = [wf.name for wf in wheel_file]
                raise KedroCliError(
                    f"More than 1 or no wheel files found: {str(file_names)}. "
                    "There has to be exactly one distribution file."
                )
            ZipFile(wheel_file[0]).extractall(temp_dir_path)

        dist_info_file = list(temp_dir_path.glob("*.dist-info"))
        if len(dist_info_file) != 1:
            raise KedroCliError(
                f"More than 1 or no dist-info files found from {package_path}. "
                "There has to be exactly one dist-info directory."
            )
        # Extract package name, based on the naming convention for wheel files
        # https://www.python.org/dev/peps/pep-0427/#file-name-convention
        package_name = dist_info_file[0].stem.split("-")[0]

        _clean_pycache(temp_dir_path)
        _install_files(package_name, temp_dir_path, env, alias)
Pull a modular pipeline package, unpack it and install the files to corresponding
    locations.
    
Pull a modular pipeline package, unpack it and install the files to corresponding
locations.
Pull a modular pipeline package unpack it and install the files to corresponding locations .
def pull_package(package_path, env, alias):
    import fsspec
    from kedro.io.core import get_protocol_and_path
    protocol, _ = get_protocol_and_path(package_path)
    filesystem = fsspec.filesystem(protocol)
    with tempfile.TemporaryDirectory() as temp_dir:
        temp_dir_path = Path(temp_dir).resolve()
        if package_path.endswith(".whl") and filesystem.exists(package_path):
            with filesystem.open(package_path) as fs_file:
                ZipFile(fs_file).extractall(temp_dir_path)
        else:
            python_call(
                "pip",
                ["download", "--no-deps", "--dest", str(temp_dir_path), package_path],
            )
            wheel_file = list(temp_dir_path.glob("*.whl"))
            if len(wheel_file) != 1:
                file_names = [wf.name for wf in wheel_file]
                raise KedroCliError(
                    f"More than 1 or no wheel files found: {str(file_names)}. "
                    "There has to be exactly one distribution file."
                )
            ZipFile(wheel_file[0]).extractall(temp_dir_path)
        dist_info_file = list(temp_dir_path.glob("*.dist-info"))
        if len(dist_info_file) != 1:
            raise KedroCliError(
                f"More than 1 or no dist-info files found from {package_path}. "
                "There has to be exactly one dist-info directory."
            )
        package_name = dist_info_file[0].stem.split("-")[0]
        _clean_pycache(temp_dir_path)
        _install_files(package_name, temp_dir_path, env, alias)
def pull_package ( package_path , env , alias ) : import fsspec from kedro . io . core import get_protocol_and_path protocol , _ = get_protocol_and_path ( package_path ) filesystem = fsspec . filesystem ( protocol ) with tempfile . TemporaryDirectory ( ) as temp_dir : temp_dir_path = Path ( temp_dir ) . resolve ( ) if package_path . endswith ( ".whl" ) and filesystem . exists ( package_path ) : with filesystem . open ( package_path ) as fs_file : ZipFile ( fs_file ) . extractall ( temp_dir_path ) else : python_call ( "pip" , [ "download" , "--no-deps" , "--dest" , str ( temp_dir_path ) , package_path ] , ) wheel_file = list ( temp_dir_path . glob ( "*.whl" ) ) if len ( wheel_file ) != 1 : file_names = [ wf . name for wf in wheel_file ] raise KedroCliError ( f"More than 1 or no wheel files found: {str(file_names)}. " "There has to be exactly one distribution file." ) ZipFile ( wheel_file [ 0 ] ) . extractall ( temp_dir_path ) dist_info_file = list ( temp_dir_path . glob ( "*.dist-info" ) ) if len ( dist_info_file ) != 1 : raise KedroCliError ( f"More than 1 or no dist-info files found from {package_path}. " "There has to be exactly one dist-info directory." ) package_name = dist_info_file [ 0 ] . stem . split ( "-" ) [ 0 ] _clean_pycache ( temp_dir_path ) _install_files ( package_name , temp_dir_path , env , alias )
Pull a modular pipeline package, unpack it and install the files to corresponding
locations.
Pull a modular pipeline package unpack it and install the files to corresponding locations .
"""Pull a modular pipeline package, unpack it and install the files to corresponding
    locations.
    """ # pylint: disable=import-outside-toplevel # `--no-deps` should fetch only one wheel file, and CLI should fail if that's # not the case. # Extract package name, based on the naming convention for wheel files # https://www.python.org/dev/peps/pep-0427/#file-name-convention
package_path
env
alias