Prefect Azure Integration

Warnings

• breaking Prefect 2.x (Orion) introduced significant architectural changes, including a new API, flow/task definition patterns using decorators (`@flow`, `@task`), and the 'Block' system for external integrations and credentials. Users migrating from Prefect 1.x will need to refactor flows and task definitions, and adopt the Block pattern for Azure resource configuration.
  Fix: Rewrite flows and tasks using Prefect 2.x decorators. Migrate old credential management to Prefect Blocks by creating and saving instances of `prefect-azure` Block types (e.g., `AzureBlobStorageCredentials`, `AzureBlobStorageContainer`) via the UI or `block.save()` methods.
• gotcha After installing `prefect-azure`, its Block types (e.g., `AzureBlobStorageContainer`, `AzureKeyVaultSecret`) must be registered with the Prefect server for them to be discoverable in the UI or by workers. Failing to do so will prevent their use or display.
  Fix: Run `prefect block register -m prefect_azure` in your environment after installation. This only needs to be done once per Prefect server instance, or whenever new Block types are added to the collection.
• gotcha For secure and robust authentication with Azure services, avoid hardcoding connection strings or secrets. `prefect-azure` components often leverage `azure-identity`'s `DefaultAzureCredential`, which can authenticate via environment variables, managed identities, Azure CLI, etc. Proper Azure RBAC permissions are crucial.
  Fix: Utilize Azure Service Principals or Managed Identities configured with appropriate Azure RBAC roles. Store secrets in Azure Key Vault and access them via `prefect-azure`'s Key Vault blocks, or use Prefect's native Secret Blocks. Ensure environment variables like `AZURE_CLIENT_ID`, `AZURE_TENANT_ID`, `AZURE_CLIENT_SECRET` (for service principals) are set, or use Managed Identity.
• gotcha When using `AzureBlobStorageContainer` (or similar `prefect-azure` Blocks) as `result_storage` for Prefect flows or tasks, the specific instance of the Block must be saved to the Prefect server (e.g., `block_instance.save("my-block-name")`) before it can be referenced by name in `@flow` or `@task` decorators.
  Fix: Ensure the `AzureBlobStorageContainer` Block (or any other Block used as `result_storage`) is saved to the Prefect server. This can be done programmatically using `block.save('block-name')` or via the Prefect UI. The `result_storage` parameter then takes the name of the saved block, e.g., `result_storage=AzureBlobStorageContainer.load("my-block-name")`.

Install

• pip install prefect-azure Base installation
• pip install "prefect-azure[blob_storage]" For Azure Blob Storage specific features
• pip install "prefect-azure[cosmos_db]" For Azure Cosmos DB specific features
• pip install "prefect-azure[ml_datastore]" For Azure ML Datastore specific features
• pip install "prefect-azure[all_extras]" Install with all additional capabilities

Imports

• AzureBlobStorageContainer

  from prefect_azure.blob_storage import AzureBlobStorageContainer

• AzureBlobStorageCredentials

  from prefect_azure import AzureBlobStorageCredentials

• blob_storage_download

  from prefect_azure.blob_storage import blob_storage_download

• AzureContainerInstanceJob

  from prefect_azure.container_instance import AzureContainerInstanceJob

Quickstart

This quickstart demonstrates how to download a file from Azure Blob Storage using the `prefect-azure` integration. It highlights the use of `AzureBlobStorageCredentials` and the `blob_storage_download` task. For actual usage, `AZURE_STORAGE_CONNECTION_STRING` should be set as an environment variable or credentials should be managed via Prefect Blocks and Azure's identity features. It also emphasizes the importance of registering Prefect Blocks for discovery.

import os
from prefect import flow
from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import blob_storage_download, AzureBlobStorageContainer

# Before running the flow, ensure the block is registered and saved (e.g., via CLI or a separate script):
# prefect block register -m prefect_azure
# From Python:
# blob_creds = AzureBlobStorageCredentials(connection_string=os.environ.get('AZURE_STORAGE_CONNECTION_STRING', ''))
# blob_creds.save(name="my-blob-creds")
# blob_container_block = AzureBlobStorageContainer(
#    container_name="my-container", 
#    credentials=blob_creds
# )
# blob_container_block.save(name="my-blob-container")

@flow
def example_blob_storage_download_flow():
    # Load credentials block (assuming it was saved with 'my-blob-creds')
    # In a real scenario, connection_string should be fetched securely (e.g., from Azure Key Vault or Prefect Secret Block)
    # or derived from environment variables/managed identity via DefaultAzureCredential.
    connection_string = os.environ.get('AZURE_STORAGE_CONNECTION_STRING', '')
    
    if not connection_string:
        print("Warning: AZURE_STORAGE_CONNECTION_STRING environment variable not set. Using dummy string.")
        # Fallback for example, in real world this would likely fail or use DefaultAzureCredential
        blob_storage_credentials = AzureBlobStorageCredentials()
    else:
        blob_storage_credentials = AzureBlobStorageCredentials(connection_string=connection_string)
    
    # If the AzureBlobStorageContainer block was saved via the UI or another script:
    # my_blob_container_block = AzureBlobStorageContainer.load("my-blob-container")
    # Instead, we create an ad-hoc one for this example's simplicity
    my_blob_container_block = AzureBlobStorageContainer(
        container_name="prefect", # Replace with your container name
        credentials=blob_storage_credentials
    )

    print(f"Attempting to download 'prefect.txt' from container 'prefect'...")
    data = blob_storage_download(
        blob="prefect.txt", # Replace with your blob name
        container="prefect", # Or use my_blob_container_block directly if it defines the container
        blob_storage_credentials=blob_storage_credentials,
    )
    print(f"Downloaded data (first 100 chars): {data[:100].decode()}")
    return data

if __name__ == "__main__":
    # Set a dummy connection string for local testing if not already set
    # For actual Azure access, replace with a real connection string or use Azure identity management
    # os.environ['AZURE_STORAGE_CONNECTION_STRING'] = 'DefaultEndpointsProtocol=https;AccountName=youraccount;AccountKey=yourkey;EndpointSuffix=core.windows.net'
    example_blob_storage_download_flow()