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>=1.0.0,<2.0.0",
 ]
 
 [project.scripts]

requirements-dev.txt

@@ -12,6 +12,7 @@ psutil==7.0.0
 requests
 cryptography
 fabric-cicd>=0.3.1
+deltalake>=1.0.0,<2.0.0
 
 # Testing and Building Requirements
 tox>=4.20.0

src/fabric_cli/client/fab_delta_client.py

@@ -0,0 +1,74 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import json
+from argparse import Namespace
+
+from deltalake import DeltaTable
+from deltalake.exceptions import DeltaError
+
+from fabric_cli.core import fab_constant
+from fabric_cli.core.fab_auth import FabAuth
+from fabric_cli.core.fab_exceptions import FabricCLIError
+from fabric_cli.errors import ErrorMessages
+
+# Item types whose OneLake Tables/ folder contains standard Delta tables.
+# SemanticModel is explicitly excluded: its Tables/ entries are columnar
+# semantic-model representations, not Delta-compatible parquet.
+_DELTA_SUPPORTED_ITEM_TYPES: frozenset[str] = frozenset(
+    {
+        "Lakehouse",
+        "Warehouse",
+        "KQLDatabase",
+        "MirroredDatabase",
+        "SQLDatabase",
+    }
+)
+
+
+def get_table_schema(args: Namespace, local_path: str) -> list[dict]:
+    """Return schema fields for a Delta table on OneLake.
+
+    Single seam for DeltaTable construction: owns item-type validation,
+    URI building, storage_options assembly, token acquisition, and
+    exception mapping.
+    """
+    item_type = getattr(args, "item_type", None)
+    if item_type is not None and item_type not in _DELTA_SUPPORTED_ITEM_TYPES:
+        raise FabricCLIError(
+            ErrorMessages.Table.unsupported_item_type_for_delta(item_type),
+            fab_constant.ERROR_INVALID_ITEM_TYPE,
+        )
+
+    token = FabAuth().get_access_token(fab_constant.SCOPE_ONELAKE_DEFAULT)
+    if token is None:
+        raise FabricCLIError(
+            ErrorMessages.Auth.access_token_failed(),
+            fab_constant.ERROR_AUTHENTICATION_FAILED,
+        )
+
+    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",
+            },
+        )
+        schema_dict = json.loads(table.schema().to_json())
+        schema_fields = schema_dict.get("fields")
+        if not isinstance(schema_fields, list):
+            raise ValueError(
+                "Delta table schema JSON does not contain a valid 'fields' list."
+            )
+        return schema_fields
+    except (DeltaError, json.JSONDecodeError, ValueError) as exc:
+        raise FabricCLIError(
+            "Failed to extract the table schema. Please ensure the path points to a valid Delta table.",
+            fab_constant.ERROR_INVALID_DELTA_TABLE,
+        ) from exc

src/fabric_cli/commands/tables/fab_tables_schema.py

@@ -1,80 +1,17 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
-import json
 from argparse import Namespace
-from typing import Optional
 
-from fabric_cli.client import fab_api_onelake as onelake_api
-from fabric_cli.core import fab_constant
-from fabric_cli.core import fab_handle_context as handle_context
-from fabric_cli.core.fab_exceptions import FabricCLIError
-from fabric_cli.core.hiearchy.fab_hiearchy import OneLakeItem
+from fabric_cli.client import fab_delta_client as delta_client
 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)
 
-    else:
-        raise FabricCLIError(
-            "Failed to extract the table schema. Please ensure the path points to a valid Delta table",
-            fab_constant.ERROR_INVALID_DETLA_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
+def _get_table_schema(args: Namespace) -> list[dict]:
+    return delta_client.get_table_schema(args, args.table_local_path)

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"]
-

src/fabric_cli/errors/table.py

@@ -13,4 +13,11 @@ def invalid_format_argument(part: str) -> str:
 
     @staticmethod
     def invalid_key(key: str, allowed_keys: str) -> str:
-        return f"Invalid key: '{key}'. Allowed keys are: {allowed_keys}"
+        return f"Invalid key: '{key}'. Allowed keys are: {allowed_keys}"
+
+    @staticmethod
+    def unsupported_item_type_for_delta(item_type: str) -> str:
+        return (
+            f"'{item_type}' does not expose Delta tables under Tables/. "
+            "Delta table schema is supported for: Lakehouse, Warehouse, KQLDatabase, MirroredDatabase, SQLDatabase."
+        )

src/fabric_cli/utils/fab_cmd_table_utils.py

@@ -7,6 +7,7 @@
 from fabric_cli.core.fab_exceptions import FabricCLIError
 from fabric_cli.core.hiearchy.fab_onelake_element import OneLakeItem
 from fabric_cli.errors import ErrorMessages
+from fabric_cli.utils import fab_util as utils
 
 
 def add_table_props_to_args(args: Any, context: OneLakeItem) -> None:
@@ -21,8 +22,9 @@ def add_table_props_to_args(args: Any, context: OneLakeItem) -> None:
     args.lakehouse_path = context.item.path
 
     table_path = context.local_path.split("/")
-    args.table_name = table_path[-1]
+    args.table_name = utils.remove_dot_suffix(table_path[-1])
     args.schema = table_path[-2] if len(table_path) == 3 else None
+    args.table_local_path = utils.remove_dot_suffix(context.local_path)
 
 
 def convert_hours_to_dhhmmss(hours: int) -> str:

tests/test_commands/commands_parser.py

@@ -2,6 +2,7 @@
 # Licensed under the MIT License.
 
 import platform
+
 from prompt_toolkit.input import DummyInput
 from prompt_toolkit.output import DummyOutput
 
@@ -12,6 +13,9 @@
 from fabric_cli.parsers.fab_config_parser import (
     register_parser as register_config_parser,
 )
+from fabric_cli.parsers.fab_find_parser import (
+    register_parser as register_find_parser,
+)
 from fabric_cli.parsers.fab_fs_parser import (
     register_assign_parser,
     register_cd_parser,
@@ -32,13 +36,13 @@
     register_stop_parser,
     register_unassign_parser,
 )
-from fabric_cli.parsers.fab_find_parser import (
-    register_parser as register_find_parser,
-)
 from fabric_cli.parsers.fab_jobs_parser import register_parser as register_jobs_parser
 from fabric_cli.parsers.fab_labels_parser import (
     register_parser as register_labels_parser,
 )
+from fabric_cli.parsers.fab_tables_parser import (
+    register_parser as register_tables_parser,
+)
 
 parserHandlers = [
     register_labels_parser,
@@ -65,6 +69,7 @@
     register_rm_parser,
     register_mkdir_parser,
     register_jobs_parser,
+    register_tables_parser,
 ]