kedro/io/core.py
def _parse_filepath(filepath: str) -> Dict[str, str]:
    """Split filepath on protocol and path. Based on `fsspec.utils.infer_storage_options`.

    Args:
        filepath: Either local absolute file path or URL (s3://bucket/file.csv)

    Returns:
        Parsed filepath.
    """
    if (
        re.match(r"^[a-zA-Z]:[\\/]", filepath)
        or re.match(r"^[a-zA-Z0-9]+://", filepath) is None
    ):
        return {"protocol": "file", "path": filepath}

    parsed_path = urlsplit(filepath)
    protocol = parsed_path.scheme or "file"

    if protocol in HTTP_PROTOCOLS:
        return {"protocol": protocol, "path": filepath}

    path = parsed_path.path
    if protocol == "file":
        windows_path = re.match(r"^/([a-zA-Z])[:|]([\\/].*)$", path)
        if windows_path:
            path = "{}:{}".format(*windows_path.groups())

    options = {"protocol": protocol, "path": path}

    if parsed_path.netloc:
        if protocol in CLOUD_PROTOCOLS:
            host_with_port = parsed_path.netloc.rsplit("@", 1)[-1]
            host = host_with_port.rsplit(":", 1)[0]
            options["path"] = host + options["path"]

    return options
Split filepath on protocol and path. Based on `fsspec.utils.infer_storage_options`.

    Args:
        filepath: Either local absolute file path or URL (s3://bucket/file.csv)

    Returns:
        Parsed filepath.
    
Split filepath on protocol and path.
Split filepath on protocol and path .
def _parse_filepath(filepath: str) -> Dict[str, str]:
    if (
        re.match(r"^[a-zA-Z]:[\\/]", filepath)
        or re.match(r"^[a-zA-Z0-9]+://", filepath) is None
    ):
        return {"protocol": "file", "path": filepath}
    parsed_path = urlsplit(filepath)
    protocol = parsed_path.scheme or "file"
    if protocol in HTTP_PROTOCOLS:
        return {"protocol": protocol, "path": filepath}
    path = parsed_path.path
    if protocol == "file":
        windows_path = re.match(r"^/([a-zA-Z])[:|]([\\/].*)$", path)
        if windows_path:
            path = "{}:{}".format(*windows_path.groups())
    options = {"protocol": protocol, "path": path}
    if parsed_path.netloc:
        if protocol in CLOUD_PROTOCOLS:
            host_with_port = parsed_path.netloc.rsplit("@", 1)[-1]
            host = host_with_port.rsplit(":", 1)[0]
            options["path"] = host + options["path"]
    return options
def _parse_filepath ( filepath : str ) -> Dict [ str , str ] : if ( re . match ( r"^[a-zA-Z]:[\\/]" , filepath ) or re . match ( r"^[a-zA-Z0-9]+://" , filepath ) is None ) : return { "protocol" : "file" , "path" : filepath } parsed_path = urlsplit ( filepath ) protocol = parsed_path . scheme or "file" if protocol in HTTP_PROTOCOLS : return { "protocol" : protocol , "path" : filepath } path = parsed_path . path if protocol == "file" : windows_path = re . match ( r"^/([a-zA-Z])[:|]([\\/].*)$" , path ) if windows_path : path = "{}:{}" . format ( * windows_path . groups ( ) ) options = { "protocol" : protocol , "path" : path } if parsed_path . netloc : if protocol in CLOUD_PROTOCOLS : host_with_port = parsed_path . netloc . rsplit ( "@" , 1 ) [ - 1 ] host = host_with_port . rsplit ( ":" , 1 ) [ 0 ] options [ "path" ] = host + options [ "path" ] return options
Split filepath on protocol and path.
Split filepath on protocol and path .
"""Split filepath on protocol and path. Based on `fsspec.utils.infer_storage_options`.

    Args:
        filepath: Either local absolute file path or URL (s3://bucket/file.csv)

    Returns:
        Parsed filepath.
    """
None
filepath
str
Either local absolute file path or URL (s3://bucket/file.csv)
Either local absolute file path or URL ( s3 : // bucket / file . csv )