Repository: kaskada-ai/kaskada
Issue: bug: file:///

**Description**
There is currently discrepancy between the usage of the Python Client and the CLI. There is various support from the API to the CLI for the file paths allowed. In the Python Client, we are not able to pass in the file URI e.g. `file://awkward.tacos.parquet` since it creates the path under the hood.

**To Reproduce**
Steps to reproduce the behavior:
1. Load data to a table using the `file://path` instead of the path.
```python
kaskada.load("table_name", "file:///Users/kevin.nguyen/Desktop/super-secret.parquet")
```

**Actual Behavior**
Throws a validation error.

**Expected Behavior**
The paths provided should be the same across the Python Client and the CLI.



Changed files:
clients/python/src/kaskada/utils.py, clients/python/tests/test_table.py

diff --git a/clients/python/src/kaskada/utils.py b/clients/python/src/kaskada/utils.py
index 254ae753..f1d23f07 100644
--- a/clients/python/src/kaskada/utils.py
+++ b/clients/python/src/kaskada/utils.py
@@ -102,14 +102,18 @@ def handleGrpcError(rpc_error: grpc.Call):
 def to_uri(file: str) -> str:
     # Remote file stores are prefixed appropriately.
     if (
-        file.startswith("s3://")
-        or file.startswith("gs://")
-        or file.startswith("https://")
+        file.startswith("s3://")  # Amazon Web Services S3 file path provided
+        or file.startswith("gs://")  # Google Cloud Platform GS file path provided
+        or file.startswith("https://")  # HTTPS file path
+        or file.startswith("file:///")  # Absolute file path is provided
     ):
         return file
-    # TODO: Verify that the object actually exists locally
-    path = Path(file).absolute()
-    return f"file://{path}"
+    elif file.startswith("file://"):  # Convert local path to full path
+        file_suffix_path = file[7:]  # The length of file://
+        return f"file://{Path(file_suffix_path).absolute()}"
+    else:  # A non-prefixed file path was provided
+        path = Path(file).absolute()
+        return f"file://{path}"
 
 
 def unpack_details(grpc_detail):
diff --git a/clients/python/tests/test_table.py b/clients/python/tests/test_table.py
index a105a7a5..9430608b 100644
--- a/clients/python/tests/test_table.py
+++ b/clients/python/tests/test_table.py
@@ -224,6 +224,42 @@ def test_table_load_csv(mockClient):
     )
 
 
+@patch("kaskada.client.Client")
+def test_table_load_csv_file_prefix(mockClient):
+    table_name = "test_table"
+    path = "local.csv"
+    local_file = f"file://{path}"
+    expected_request = table_pb.LoadDataRequest(
+        table_name=table_name,
+        file_input=common_pb.FileInput(
+            file_type="FILE_TYPE_CSV", uri=f"file://{Path(path).absolute()}"
+        ),
+    )
+
+    kaskada.table.load(table_name, local_file, client=mockClient)
+    mockClient.table_stub.LoadData.assert_called_with(
+        expected_request, metadata=mockClient.get_metadata()
+    )
+
+
+@patch("kaskada.client.Client")
+def test_table_load_from_absolute_path(mockClient):
+    table_name = "test_table"
+    path = "local.csv"
+    local_file = f"file:///{path}"
+    expected_request = table_pb.LoadDataRequest(
+        table_name=table_name,
+        file_input=common_pb.FileInput(
+            file_type="FILE_TYPE_CSV", uri=f"file:///{path}"
+        ),
+    )
+
+    kaskada.table.load(table_name, local_file, client=mockClient)
+    mockClient.table_stub.LoadData.assert_called_with(
+        expected_request, metadata=mockClient.get_metadata()
+    )
+
+
 @patch("kaskada.client.Client")
 def test_table_load_csv_azure(mockClient):
     table_name = "test_table"