Environments¶

Fabric Environments define the Spark runtime, system libraries, and custom packages for notebooks and jobs. The Environment API provides management of environments and their associated libraries.

Getting an Environment¶

import fabias

ws = fabias.workspace("Analytics")

# By name
env = ws.environment("ML Environment")

# By GUID
env = ws.environment("a1b2c3d4-...")

Listing Environments¶

ws = fabias.workspace("Analytics")

# List all environments
for env in ws.environments():
    print(f"{env.name}: {env.id}")

Creating Environments¶

ws = fabias.workspace("Analytics")

# Create a new environment
env = ws.environments.add("Data Science")

# Create with description and folder
env = ws.environments.add(
    "Data Science",
    description="Python environment for data science workloads",
    folder_id="folder-guid"
)

Environment Properties¶

env = ws.environment("ML Environment")

# Refresh to get latest state
env.refresh()

print(f"ID: {env.id}")
print(f"Name: {env.name}")
print(f"State: {env.state}")
print(f"Is Publishing: {env.isPublishing}")

Publish States¶

The state property can have these values:

State	Description
`Running`	Currently publishing
`Waiting`	Queued for publishing
`Cancelling`	Cancellation in progress
`Success`	Published successfully
`Failed`	Publishing failed

Managing Libraries¶

Environments support custom Python, R, and Java libraries via the libraries property:

List Libraries¶

env = ws.environment("ML Environment")
libs = env.libraries

# Refresh library data
libs.refresh()

# View staging libraries (pending publish)
print("Staging Libraries:")
for lib in libs.staging["libraries"]:
    print(f"  - {lib}")

# View published libraries (currently active)
print("\nPublished Libraries:")
for lib in libs.published["libraries"]:
    print(f"  - {lib}")

Upload Libraries¶

Supports .whl (Python wheel), .py (Python script), .jar (Java), and .tar.gz (R) files:

env = ws.environment("ML Environment")
libs = env.libraries

# Upload a wheel file
libs.upload("/path/to/my_package-1.0.0-py3-none-any.whl")

# Upload a Python script
libs.upload("/path/to/helpers.py")

# Upload a JAR file
libs.upload("/path/to/spark-connector.jar")

Replace Existing Libraries¶

Replace an existing library with a new version:

env = ws.environment("ML Environment")
libs = env.libraries

# Replace with callback for progress tracking
def progress(action, details):
    print(f"{action}: {details}")

libs.replace(
    "my_package",  # Regex pattern to match library name
    "/path/to/my_package-2.0.0-py3-none-any.whl",
    callback=progress
)

The replace() method: 1. Lists staging libraries 2. Deletes all matching libraries (by regex pattern) 3. Uploads the new library file

Delete Libraries¶

env = ws.environment("ML Environment")
libs = env.libraries

# Delete from staging
libs.delete("my_package-1.0.0-py3-none-any.whl")

Publishing Changes¶

After uploading or deleting libraries, publish the environment to make changes active:

env = ws.environment("ML Environment")
libs = env.libraries

# Upload a new library
libs.upload("/path/to/new_package.whl")

# Publish the environment
env.publish()

# Optional: Wait for publish to complete
env.waitForPublish(
    callback=lambda e: print(f"State: {e.state}"),
    timeout=1800,  # 30 minutes
    poll_interval=30  # Check every 30 seconds
)

print("Environment published successfully!")

Cancel Publishing¶

If an environment is publishing and you need to cancel:

env = ws.environment("ML Environment")

if env.isPublishing:
    env.cancelPublish()
    print("Publish cancelled")

Example: Add Package and Publish¶

import fabias

ws = fabias.workspace("Analytics")
env = ws.environment("ML Environment")

# Upload a custom package
print("Uploading library...")
env.libraries.upload("/path/to/my_ml_utils-1.0.0-py3-none-any.whl")

# Publish the environment
print("Publishing environment...")
result = env.publish()

# Wait for completion with progress updates
env.waitForPublish(
    callback=lambda e: print(f"  Status: {e.state}"),
    timeout=1200,  # 20 minutes
    poll_interval=20
)

print("? Environment is ready!")

Example: Library Upgrade Workflow¶

import fabias

def upgrade_library(workspace_name: str, env_name: str, 
                    old_pattern: str, new_file: str):
    """Upgrade a library in an environment."""

    ws = fabias.workspace(workspace_name)
    env = ws.environment(env_name)
    libs = env.libraries

    print(f"Upgrading {old_pattern} in {env_name}...")

    # Replace old version with new
    libs.replace(
        old_pattern,
        new_file,
        callback=lambda action, details: print(f"  {action}: {details}")
    )

    # Publish changes
    print("Publishing...")
    env.publish()

    # Wait for completion
    env.waitForPublish(
        callback=lambda e: print(f"  {e.state}"),
        timeout=1800
    )

    print("? Upgrade complete!")

# Usage
upgrade_library(
    "Analytics",
    "ML Environment",
    "pandas",
    "/path/to/pandas-2.2.0-cp311-cp311-manylinux_2_17_x86_64.whl"
)

Example: Compare Staging vs Published¶

import fabias

ws = fabias.workspace("Analytics")
env = ws.environment("ML Environment")
libs = env.libraries

# Refresh library data
libs.refresh()

staging_set = set(libs.staging["libraries"])
published_set = set(libs.published["libraries"])

# New libraries in staging
new_libs = staging_set - published_set
if new_libs:
    print("New libraries to be published:")
    for lib in new_libs:
        print(f"  + {lib}")

# Removed libraries
removed_libs = published_set - staging_set
if removed_libs:
    print("\nLibraries to be removed:")
    for lib in removed_libs:
        print(f"  - {lib}")

# Unchanged
if not new_libs and not removed_libs:
    print("No changes in staging")

Refreshing Environment Data¶

Environment properties and library data are cached. To force a refresh:

env = ws.environment("ML Environment")

# Refresh environment metadata
env.refresh()
print(f"State: {env.state}")

# Refresh library lists
libs = env.libraries
libs.refresh()
print(f"Staging libraries: {len(libs.staging['libraries'])}")

Error Handling¶

Publish Failures¶

from fabias import fabiasError

env = ws.environment("ML Environment")

try:
    env.publish()
    env.waitForPublish(timeout=1800)
except FabiasError as e:
    print(f"Publish failed: {e}")

    # Check final state
    env.refresh()
    print(f"Final state: {env.state}")

Upload Failures¶

from fabias import fabiasError

try:
    env.libraries.upload("/path/to/invalid_file.txt")
except FabiasError as e:
    print(f"Upload failed: {e}")