feat: add support for environment definitions

.changes/unreleased/added-20260417-105920.yaml

@@ -0,0 +1,6 @@
+kind: added
+body: Environment definitions support
+time: 2026-04-17T10:59:20.288002+03:00
+custom:
+    Author: v-alexmoraru
+    AuthorLink: https://github.com/v-alexmoraru

src/fabric_cli/commands/fs/impor/fab_fs_import_item.py

@@ -8,7 +8,6 @@
 from fabric_cli.client.fab_api_types import ApiResponse
 from fabric_cli.core import fab_constant, fab_logger
 from fabric_cli.core.fab_exceptions import FabricCLIError
-from fabric_cli.core.fab_types import ItemType
 from fabric_cli.core.hiearchy.fab_hiearchy import Item
 from fabric_cli.utils import fab_cmd_import_utils as utils_import
 from fabric_cli.utils import fab_item_util
@@ -53,11 +52,7 @@ def import_single_item(item: Item, args: Namespace) -> None:
                     f"Importing (update) '{_input_path}' → '{item.path}'..."
                 )
 
-                # Environment item type, not supporting definition yet
-                if item.item_type == ItemType.ENVIRONMENT:
-                    _import_update_environment_item(args, payload)
-                else:
-                    _import_update_item(args, payload)
+                _import_update_item(args, payload)
 
                 utils_ui.print_output_format(
                     args, message=f"'{item.name}' imported")
@@ -66,11 +61,7 @@ def import_single_item(item: Item, args: Namespace) -> None:
             utils_ui.print_grey(
                 f"Importing '{_input_path}' → '{item.path}'...")
 
-            # Environment item type, not supporting definition yet
-            if item.item_type == ItemType.ENVIRONMENT:
-                response = _import_create_environment_item(item, args, payload)
-            else:
-                response = _import_create_item(args, payload)
+            response = _import_create_item(args, payload)
 
             if response.status_code in (200, 201):
                 utils_ui.print_output_format(
@@ -83,10 +74,6 @@ def import_single_item(item: Item, args: Namespace) -> None:
 
 
 # Utils
-def _import_update_environment_item(args: Namespace, payload: dict) -> None:
-    utils_import.publish_environment_item(args, payload)
-
-
 def _import_update_item(args: Namespace, payload: dict) -> None:
     definition_payload = json.dumps(
         {
@@ -96,27 +83,6 @@ def _import_update_item(args: Namespace, payload: dict) -> None:
     item_api.update_item_definition(args, payload=definition_payload)
 
 
-def _import_create_environment_item(
-    item: Item, args: Namespace, payload: dict
-) -> ApiResponse:
-
-    item_payload: dict = {
-        "type": str(item.item_type),
-        "description": "Imported from fab",
-        "displayName": item.short_name,
-        "folderId": item.folder_id,
-    }
-    item_payload_str = json.dumps(item_payload)
-
-    # Create the item
-    response = item_api.create_item(args, payload=item_payload_str)
-    data = json.loads(response.text)
-    args.id = data["id"]
-
-    utils_import.publish_environment_item(args, payload)
-    return response
-
-
 def _import_create_item(args: Namespace, payload: dict) -> ApiResponse:
     _payload = json.dumps(payload)
     return item_api.create_item(args, payload=_payload)

src/fabric_cli/core/fab_config/command_support.yaml

@@ -256,6 +256,7 @@ commands:
           - user_data_function
           - graph_query_set
           - map
+          - environment
           - lakehouse
       import:
         supported_items:

src/fabric_cli/core/fab_types.py

@@ -582,6 +582,7 @@ class MirroredDatabaseFolders(Enum):
 # Item Payload definition
 
 definition_format_mapping = {
+    ItemType.ENVIRONMENT: {"default": ""},
     ItemType.SPARK_JOB_DEFINITION: {
         "default": "SparkJobDefinitionV1",
         "SparkJobDefinitionV1": "SparkJobDefinitionV1",

src/fabric_cli/utils/fab_cmd_import_utils.py

@@ -4,35 +4,24 @@
 import base64
 import json
 import os
-import time
-from argparse import Namespace
 from typing import Any, Optional
 
-import yaml
-
-from fabric_cli.client import fab_api_item as item_api
 from fabric_cli.core import fab_constant
 from fabric_cli.core.fab_exceptions import FabricCLIError
-from fabric_cli.core.fab_types import ItemType
 from fabric_cli.core.hiearchy.fab_hiearchy import Item
-from fabric_cli.utils import fab_ui as utils_ui
 
 
 def get_payload_for_item_type(
     path: str, item: Item, input_format: Optional[str] = None
 ) -> dict:
-    # Environment does not support updateDefinition yet, custom payload / dev
-    if item.item_type == ItemType.ENVIRONMENT:
-        return _build_environment_payload(path)
-    else:
-        definition = _build_definition(path, input_format)
-        return {
-            "type": str(item.item_type),
-            "description": "Imported from fab",
-            "folderId": item.folder_id,
-            "displayName": item.short_name,
-            "definition": definition,
-        }
+    definition = _build_definition(path, input_format)
+    return {
+        "type": str(item.item_type),
+        "description": "Imported from fab",
+        "folderId": item.folder_id,
+        "displayName": item.short_name,
+        "definition": definition,
+    }
 
 
 def _build_definition(input_path: Any, input_format: Optional[str] = None) -> dict:
@@ -82,218 +71,3 @@ def _build_definition(input_path: Any, input_format: Optional[str] = None) -> di
 def _encode_file_to_base64(file_path: str) -> str:
     with open(file_path, "rb") as file:
         return base64.b64encode(file.read()).decode("utf-8")
-
-
-# Environments
-
-
-def publish_environment_item(args: Namespace, payload: dict) -> None:
-    # Check for ongoing publish
-    _check_environment_publish_state(args, True)
-
-    # Update compute settings
-    _update_compute_settings(args, payload)
-
-    # Add libraries to environment, overwriting anything with the same name and return the list of libraries
-    _add_libraries(args, payload)
-
-    # Remove libraries from live environment
-    _remove_libraries(args, payload)
-
-    # Publish
-    item_api.environment_publish(args)
-
-    # Wait for ongoing publish to complete
-    _check_environment_publish_state(args)
-
-    utils_ui.print_info(f"Published")
-
-
-def _check_environment_publish_state(
-    args: Namespace, initial_check: bool = False
-) -> None:
-    publishing = True
-    iteration = 1
-
-    while publishing:
-        args.item_uri = "environments"
-        response = item_api.get_item(args, item_uri=True)
-        data = response.json()
-
-        current_state = (
-            data.get("properties", {})
-            .get("publishDetails", {})
-            .get("state", "Unknown")
-            .lower()
-        )
-
-        if initial_check:
-
-            prepend_message = "Existing Environment publish is in progess"
-            pass_values = ["success", "failed", "cancelled"]
-            fail_values = []
-
-        else:
-            prepend_message = "Operation in progress"
-            pass_values = ["success"]
-            fail_values = ["failed", "cancelled"]
-
-        if current_state in pass_values:
-            publishing = False
-        elif current_state in fail_values:
-            msg = f"Publish {current_state} for Libraries"
-            raise Exception(msg)
-        else:
-            _handle_retry(
-                attempt=iteration,
-                base_delay=5,
-                max_retries=20,
-                response_retry_after=120,
-                prepend_message=prepend_message,
-            )
-            iteration += 1
-
-
-def _build_environment_payload(input_path: Any) -> dict:
-    directory = input_path
-
-    parts: dict[Any, Any] = {}
-    for root, dirs, files in os.walk(directory):
-        for file in files:
-            # Get full path and relative path
-            full_path = os.path.join(root, file)
-
-            # Spark compute settings
-            if "Setting" in full_path:
-                with open(full_path, "r") as file:
-                    yaml_body = yaml.safe_load(file)
-                parts["sparkCompute"] = _convert_environment_compute_to_camel(yaml_body)
-
-            # Spark libraries
-            elif "Libraries" in full_path:
-                parts["libraries"] = parts.get("libraries", [])
-                # Append instead of overwrite
-                parts["libraries"].append(full_path)
-
-    return {"parts": parts}
-
-
-def _convert_environment_compute_to_camel(input_dict: dict) -> dict:
-    new_input_dict = {}
-
-    for key, value in input_dict.items():
-        if key == "spark_conf":
-            new_key = "sparkProperties"
-        else:
-            # Convert the key to camelCase
-            key_components = key.split("_")
-            # Capitalize the first letter of each component except the first one
-            new_key = key_components[0] + "".join(x.title() for x in key_components[1:])
-
-        # Recursively update dictionary values if they are dictionaries
-        if isinstance(value, dict):
-            value = _convert_environment_compute_to_camel(value)
-
-        new_input_dict[new_key] = value
-
-    return new_input_dict
-
-
-def _update_compute_settings(args: Namespace, payload: dict) -> None:
-    if "sparkCompute" in payload["parts"]:
-        spark_compute = payload["parts"]["sparkCompute"]
-        _spark_compute_payload = json.dumps(spark_compute)
-
-        args.ext_uri = "/staging/sparkcompute"
-        args.item_uri = "environments"
-
-        response = item_api.update_item(
-            args, payload=_spark_compute_payload, item_uri=True, ext_uri=True
-        )
-
-        if response.status_code == 200:
-            utils_ui.print_info("Updated Spark Settings")
-
-
-def _add_libraries(args: Namespace, payload: dict) -> None:
-    if "libraries" in payload["parts"]:
-        # Extract the list of libraries
-        library_paths = payload["parts"]["libraries"]
-
-        for file_path in library_paths:
-            file_name = os.path.basename(file_path)
-
-            # Open the file in binary mode for reading
-            with open(file_path, "rb") as file:
-                library_file = {"file": (file_name, file)}
-
-                # Upload libraries to the environment
-                response = item_api.environment_upload_staging_library(
-                    args, library_file
-                )
-
-                if response.status_code == 200:
-                    utils_ui.print_info(f"Updated Library '{file_name}'")
-
-
-def _remove_libraries(args: Namespace, payload: dict) -> None:
-    args.ext_uri = "/libraries"
-    args.item_uri = "environments"
-
-    try:
-        response = item_api.get_item(args, item_uri=True, ext_uri=True)
-        if response.status_code == 200:
-            response_json = response.json()  # Convert to dictionary
-
-            repo_library_files = tuple(
-                os.path.basename(file) for file in payload["parts"]["libraries"]
-            )
-
-            if (
-                "environmentYml" in response_json
-                and response_json["environmentYml"]  # Not None or ''
-                and "environment.yml" not in repo_library_files
-            ):
-                _remove_library(args, "environment.yml")
-
-            custom_libraries = response_json.get("customLibraries", {})
-            if isinstance(custom_libraries, dict):
-                for files in custom_libraries.values():
-                    if isinstance(files, list):
-                        for file in files:
-                            if file not in repo_library_files:
-                                _remove_library(args, file)
-
-    except Exception as e:
-        pass
-
-
-def _remove_library(args: Namespace, file_name: str) -> None:
-    item_api.environment_delete_library_staging(args, file_name)
-    utils_ui.print_info(f"Removed {file_name}")
-
-
-def _handle_retry(
-    attempt: int,
-    base_delay: float,
-    max_retries: int,
-    response_retry_after: float = 60,
-    prepend_message: str = "",
-) -> None:
-    if attempt < max_retries:
-        retry_after = float(response_retry_after)
-        base_delay = float(base_delay)
-        delay = min(retry_after, base_delay * (2**attempt))
-
-        # Modify output for proper plurality and formatting
-        delay_str = f"{delay:.0f}" if delay.is_integer() else f"{delay:.2f}"
-        second_str = "second" if delay == 1 else "seconds"
-        prepend_message += " " if prepend_message else ""
-
-        utils_ui.print_progress(
-            f"{prepend_message}Checking again in {delay_str} {second_str} (Attempt {attempt}/{max_retries})..."
-        )
-        time.sleep(delay)
-    else:
-        msg = f"Maximum retry attempts ({max_retries}) exceeded"
-        raise Exception(msg)

tests/test_commands/conftest.py

@@ -80,7 +80,8 @@
     ItemType.REPORT, ItemType.SEMANTIC_MODEL, ItemType.KQL_DATABASE,
     ItemType.KQL_QUERYSET, ItemType.EVENTHOUSE, ItemType.MIRRORED_DATABASE,
     ItemType.REFLEX, ItemType.KQL_DASHBOARD, ItemType.SQL_DATABASE,
-    ItemType.COSMOS_DB_DATABASE, ItemType.USER_DATA_FUNCTION, ItemType.LAKEHOUSE
+    ItemType.COSMOS_DB_DATABASE, ItemType.USER_DATA_FUNCTION,
+    ItemType.ENVIRONMENT, ItemType.LAKEHOUSE,
 ])
 
 import_create_new_item_fail_params = pytest.mark.parametrize("item_type", [
@@ -161,7 +162,7 @@
     (ItemType.NOTEBOOK, True),
     (ItemType.DATA_PIPELINE, True),
     (ItemType.LAKEHOUSE, True),
-    (ItemType.ENVIRONMENT, False),
+    (ItemType.ENVIRONMENT, True),
     (ItemType.WAREHOUSE, False),
     (ItemType.COSMOS_DB_DATABASE, True),
     (ItemType.USER_DATA_FUNCTION, True),
@@ -252,6 +253,7 @@
     ItemType.NOTEBOOK,
     ItemType.SPARK_JOB_DEFINITION,
     ItemType.DATA_PIPELINE,
+    ItemType.ENVIRONMENT,
     ItemType.MIRRORED_DATABASE,
     ItemType.REPORT,
     ItemType.SEMANTIC_MODEL,