Automating Microsoft Fabric Environment Creation with Custom Python Packages from Azure DevOps Artifacts

As data engineers and analysts, we often find ourselves juggling multiple tools and platforms. Microsoft Fabric has emerged as a powerful unified analytics platform, but setting up environments with custom packages can be a time-consuming task. In this article, I’ll walk you through a Python script I developed to automate the creation and updating of Microsoft Fabric environments, complete with custom Python packages from Azure DevOps Artifacts.

The Challenge

Manually creating Fabric environments and installing custom packages is tedious and prone to errors. Moreover, when working in teams, ensuring consistency across different environments becomes crucial. That’s where automation comes in handy.

The Solution

I’ve created a Python class called UploadPackageToEnvironment that handles the entire process:

1. Creating a new Fabric environment (or updating an existing one)
2. Downloading a custom package from Azure DevOps Artifacts
3. Uploading the package to the Fabric environment
4. Publishing the environment to make changes effective

Let’s break down the key components of this solution.

Setting Up the Environment

First, we need to initialize our UploadPackageToEnvironment class with the necessary parameters:

uploader = UploadPackageToEnvironment(
    environment="MyFabricEnv",
    workspace_name="MyWorkspace",
    fabric_access_token="your_fabric_token",
    is_devops=True,
    package_name="my_custom_package",
    package_version="1.0.0",
    devops_pat="your_devops_pat"
)

This setup allows us to specify whether we’re pulling a package from Azure DevOps or an external source.

Creating the Fabric Environment

The create_fabric_item method handles the creation of our Fabric environment. This method first checks if the environment already exists. If it does, we’ll use the existing environment. If not, it creates a new one.

def create_fabric_item(self, payload: dict, workspace=None):
    """
    Create a fabric item (lakehouse, env) using API POST request
    Params:
        payload (dict) : Request payload. Example payload = {"displayName": "TestingEnvAPI",
                        "type": "Environment",
                        description": "Default environment in Microsoft Fabric in which to  custom library and required packages"}

        workspace (str): Name of the workspace where to create the fabric item
    """
    workspaceId = (
        fabric.get_workspace_id()
        if workspace is None
        else fabric.resolve_workspace_id(workspace)
    )
    environment_id = self.check_if_environment_exist(
        payload.get("displayName"), workspaceId
    )

    if environment_id:
        print(
            f"Environment {payload.get('displayName')} already exist. Moving to upload step"
        )
        return environment_id, workspaceId
    try:
        response = self.fabric_client.post(
            f"/v1/workspaces/{workspaceId}/items", json=payload
        )

        if response.status_code != 201:
            raise FabricHTTPException(response)
        environment_id = response.json()["id"]
        workspace_id = response.json()["workspaceId"]
        print(
            f"Environment {payload.get('displayName')} succesfully created. Moving to upload step"
        )
        return environment_id, workspace_id
    except WorkspaceNotFoundException as e:
        raise ("Caught a WorkspaceNotFoundException:", e)
    except FabricHTTPException as e:
        raise ("Caught a FabricHTTPException. Check the API endpoint, authentication.")
    return

Downloading the Package

For Azure DevOps packages, we use the download_package_from_azure_devops method. This method constructs the correct URL for the Azure DevOps Artifacts feed, authenticates with the provided token, and downloads the specified package version.

def download_package_from_azure_devops(self, token, package_name, package_version):
    """
    Function to get download .whl file from Azure Artifact Feed
    Params:
        token (str): Azure DevOps PAT
        package_name (str): Name of the package
        package_version (str): Package version
    """
    base_url = "https://pkgs.dev.azure.com/OrganizationName/ProjectName/_packaging/FeedName/pypi/simple"

    package_url = f"{base_url}/{package_name}/"
    auth = HTTPBasicAuth("", token)
    response = requests.get(package_url, auth=auth)
    soup = BeautifulSoup(response.text, "html.parser")
    download_url = soup.find_all("a", href=True)[0]["href"]
    filename = self.get_package_whl(download_url, auth=auth, is_devops=True)
    return filename

Uploading to Fabric

Once we have our package, we upload it to the Fabric environment using the Fabric environment APIs

def upload_package_to_fabric(
    self, workspace_id, environment_id, file_path, fabric_access_token
):
    """
    Upload a custom python package to fabric environment
    Params:
        workspace_id (str): The idea of the workspace where to uoload the library
        environment_id (str): The environment id within the workspace
        file_path (str): The downloaded .whl file from Atifact feeds or
    """
    url = f"{self.base_fabric_url}/workspaces/{workspace_id}/environments/{environment_id}/staging/libraries"

    with open(file_path, "rb") as file:
        file_content = file.read()
    m = MultipartEncoder(
        fields={
            "file": (
                os.path.basename(file_path),
                file_content,
                "application/octet-stream",
            )
        }
    )

    headers = {
        "Content-Type": m.content_type,
        "Authorization": f"Bearer {fabric_access_token}",
    }

    response = requests.post(url, headers=headers, data=m)

    if response.status_code == 200:
        print("Package uploaded successfully to Fabric environment")
    else:
        print(
            f"Failed to upload package to Fabric. Status code: {response.status_code}"
        )
        print(f"Response: {response.text}")

Publishing the Environment

Finally, we publish the environment to make our changes effective:

def publish_environment(self, workspace_id, environment_id):
    url = f"{self.base_fabric_url}/workspaces/{workspace_id}/environments/{environment_id}/staging/publish"
    response = requests.post(url, headers=self.headers)
    if response.status_code == 200:
        return response.json()
    else:
        print(f"Failed to publish environment. Status code: {response.status_code}")
        return None

The wait_for_publish_completion method then polls the API to ensure our publish operation completes successfully.

Putting It All Together

With all these pieces in place, creating or updating a Fabric environment with a custom package is as simple as:

uploader = UploadPackageToEnvironment(...)
uploader.create_and_upload()

Conclusion

Automating the creation and updating of Microsoft Fabric environments with custom packages can significantly streamline your workflow. This script provides a flexible, reusable solution that can be easily integrated into your CI/CD pipelines or run on-demand.

By leveraging the APIs provided by Microsoft Fabric and Azure DevOps, you can create a robust tool that saves time, reduces errors, and ensures consistency across environments.

Note

Remember to handle your access tokens securely and consider implementing additional error handling and logging for production use. Happy automating!

To get the full Code for the complete implementation, please visit the GitHub repository. Feel free to star the repo if you find it useful, and don’t hesitate to open an issue if you have any questions or suggestions for improvement!

data-engineering-toolbox/src/fabric_env_automation.py at main · BirdiD/data-engineering-toolbox (github.com)