fix: tables schema command replace manual _delta_log JSON parsing with DeltaTable.schema() from the deltalake library

.changes/unreleased/fixed-20260430-130558.yaml

@@ -0,0 +1,6 @@
+kind: fixed
+body: Refactor `fab tables schema` to use the `deltalake` Python library for schema extraction via ABFSS URI instead of manually parsing Delta log commit files
+time: 2026-04-30T13:05:58.364670843Z
+custom:
+    Author: pkontek
+    AuthorLink: https://github.com/pkontek

pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
     "requests",
     "cryptography",
     "fabric-cicd>=0.3.1",
+    "deltalake>=0.18.0",
 ]
 
 [project.scripts]

src/fabric_cli/commands/tables/fab_tables_schema.py

@@ -3,78 +3,45 @@
 
 import json
 from argparse import Namespace
-from typing import Optional
 
-from fabric_cli.client import fab_api_onelake as onelake_api
+from deltalake import DeltaTable
+from deltalake.exceptions import TableNotFoundError
+
 from fabric_cli.core import fab_constant
-from fabric_cli.core import fab_handle_context as handle_context
+from fabric_cli.core.fab_auth import FabAuth
 from fabric_cli.core.fab_exceptions import FabricCLIError
-from fabric_cli.core.hiearchy.fab_hiearchy import OneLakeItem
 from fabric_cli.utils import fab_ui
-from fabric_cli.utils import fab_util as utils
 
 
 def exec_command(args: Namespace) -> None:
-    schema = _extract_schema_from_commit_logs(args)
-    if schema:
-        fab_ui.print_grey("Schema extracted successfully")
-        _schema = json.loads(schema)["fields"]
-        fab_ui.print_output_format(args, data=_schema, show_headers=True)
+    schema_fields = _get_table_schema(args)
+    fab_ui.print_grey("Schema extracted successfully")
+    fab_ui.print_output_format(args, data=schema_fields, show_headers=True)
+
 
+def _get_table_schema(args: Namespace) -> list[dict]:
+    token = FabAuth().get_access_token(fab_constant.SCOPE_ONELAKE_DEFAULT)
+    if args.schema:
+        local_path = f"Tables/{args.schema}/{args.table_name}"
     else:
+        local_path = f"Tables/{args.table_name}"
+
+    table_uri = (
+        f"abfss://{args.ws_id}@{fab_constant.API_ENDPOINT_ONELAKE}"
+        f"/{args.lakehouse_id}/{local_path}"
+    )
+
+    try:
+        table = DeltaTable(
+            table_uri,
+            storage_options={
+                "bearer_token": token,
+                "use_fabric_endpoint": "true",
+            },
+        )
+        return table.schema().json()["fields"]
+    except TableNotFoundError:
         raise FabricCLIError(
             "Failed to extract the table schema. Please ensure the path points to a valid Delta table",
-            fab_constant.ERROR_INVALID_DETLA_TABLE,
+            fab_constant.ERROR_INVALID_DELTA_TABLE,
         )
-
-
-def _get_commit_logs(args: Namespace) -> Optional[list[str]]:
-    _delta_log_path = args.path
-    _delta_log_path[-1] = _delta_log_path[-1] + "/_delta_log"
-
-    _context = handle_context.get_command_context(_delta_log_path, raise_error=True)
-    assert isinstance(_context, OneLakeItem)
-    onelake: OneLakeItem = _context
-    workspace_id = onelake.workspace.id
-    item_id = onelake.item.id
-    local_path = onelake.local_path
-
-    local_path = utils.remove_dot_suffix(local_path)
-    args.directory = f"{workspace_id}/?recursive=false&resource=filesystem&directory={item_id}/{local_path}&getShortcutMetadata=true"
-    response = onelake_api.list_tables_files_recursive(args)
-
-    if response.status_code in {200, 201}:
-        file_names = [f["name"] for f in response.json().get("paths", [])]
-        json_files = [
-            f"{workspace_id}/{item_id}/{f.split('/', 1)[1]}"
-            for f in file_names
-            if f.endswith(".json") and f != "_temporary"
-        ]
-        json_files.sort(reverse=True)
-        return json_files
-    return None
-
-
-def _extract_schema_from_commit_logs(args: Namespace) -> Optional[str]:
-    commit_logs = _get_commit_logs(args)
-
-    if not commit_logs:
-        return None
-
-    for log in commit_logs:
-        args.from_path = log
-        args.wait = True
-        response = onelake_api.read(args)
-
-        if response.status_code in {200, 201}:
-            json_string = response.text
-            json_objects = json_string.strip().split("\n")
-
-            for obj in json_objects:
-                commit_data = json.loads(obj)
-                if "metaData" in commit_data:
-                    metadata = commit_data["metaData"]
-                    schema = metadata["schemaString"]
-                    return schema
-
-    return None

src/fabric_cli/core/fab_constant.py

@@ -19,8 +19,7 @@
 )
 
 API_ENDPOINT_POWER_BI = (
-    validate_and_get_env_variable(
-        "FAB_API_ENDPOINT_POWER_BI", "api.powerbi.com")
+    validate_and_get_env_variable("FAB_API_ENDPOINT_POWER_BI", "api.powerbi.com")
     + "/v1.0/myorg"
 )
 
@@ -264,7 +263,7 @@
 ERROR_INVALID_OPERATION = "InvalidOperation"
 ERROR_INVALID_PATH = "InvalidPath"
 ERROR_INVALID_PROPERTY = "InvalidProperty"
-ERROR_INVALID_DETLA_TABLE = "InvalidDeltaTable"
+ERROR_INVALID_DELTA_TABLE = "InvalidDeltaTable"
 ERROR_INVALID_QUERY_FIELDS = "InvalidQueryFields"
 ERROR_INVALID_WORKSPACE_TYPE = "InvalidWorkspaceType"
 ERROR_INVALID_QUERY = "InvalidQuery"
@@ -351,4 +350,3 @@
 
 # Invalid query parameters for set command across all fabric resources
 SET_COMMAND_INVALID_QUERIES = ["id", "type", "workspaceId", "folderId"]
-