{"id":6415,"library":"prefect-azure","title":"Prefect Azure Integration","description":"prefect-azure provides a collection of Prefect integrations for orchestrating workflows with Microsoft Azure services. It enables interaction with Azure Blob Storage, Azure Key Vault, Azure Container Instances (ACI), and more. The library is actively maintained, with releases typically aligning with the Prefect core development cycle, and the current version is 0.4.9.","status":"active","version":"0.4.9","language":"en","source_language":"en","source_url":"https://github.com/PrefectHQ/prefect/tree/main/src/integrations/prefect-azure","tags":["Prefect","Azure","Cloud","Orchestration","Data Workflow","Blob Storage","Container Instances","Key Vault"],"install":[{"cmd":"pip install prefect-azure","lang":"bash","label":"Base installation"},{"cmd":"pip install \"prefect-azure[blob_storage]\"","lang":"bash","label":"For Azure Blob Storage specific features"},{"cmd":"pip install \"prefect-azure[cosmos_db]\"","lang":"bash","label":"For Azure Cosmos DB specific features"},{"cmd":"pip install \"prefect-azure[ml_datastore]\"","lang":"bash","label":"For Azure ML Datastore specific features"},{"cmd":"pip install \"prefect-azure[all_extras]\"","lang":"bash","label":"Install with all additional capabilities"}],"dependencies":[{"reason":"Core dependency for Prefect workflow orchestration.","package":"prefect","optional":false},{"reason":"Common dependency for Azure SDK authentication patterns (e.g., DefaultAzureCredential).","package":"azure-identity","optional":false},{"reason":"Required for Azure Blob Storage features (part of `[blob_storage]` extra).","package":"azure-storage-blob","optional":true},{"reason":"Required for Azure Cosmos DB features (part of `[cosmos_db]` extra).","package":"azure-cosmos","optional":true},{"reason":"Required for Azure ML Datastore features (part of `[ml_datastore]` extra).","package":"azureml-fsspec","optional":true}],"imports":[{"symbol":"AzureBlobStorageContainer","correct":"from prefect_azure.blob_storage import AzureBlobStorageContainer"},{"symbol":"AzureBlobStorageCredentials","correct":"from prefect_azure import AzureBlobStorageCredentials"},{"symbol":"blob_storage_download","correct":"from prefect_azure.blob_storage import blob_storage_download"},{"symbol":"AzureContainerInstanceJob","correct":"from prefect_azure.container_instance import AzureContainerInstanceJob"}],"quickstart":{"code":"import os\nfrom prefect import flow\nfrom prefect_azure import AzureBlobStorageCredentials\nfrom prefect_azure.blob_storage import blob_storage_download, AzureBlobStorageContainer\n\n# Before running the flow, ensure the block is registered and saved (e.g., via CLI or a separate script):\n# prefect block register -m prefect_azure\n# From Python:\n# blob_creds = AzureBlobStorageCredentials(connection_string=os.environ.get('AZURE_STORAGE_CONNECTION_STRING', ''))\n# blob_creds.save(name=\"my-blob-creds\")\n# blob_container_block = AzureBlobStorageContainer(\n#    container_name=\"my-container\", \n#    credentials=blob_creds\n# )\n# blob_container_block.save(name=\"my-blob-container\")\n\n@flow\ndef example_blob_storage_download_flow():\n    # Load credentials block (assuming it was saved with 'my-blob-creds')\n    # In a real scenario, connection_string should be fetched securely (e.g., from Azure Key Vault or Prefect Secret Block)\n    # or derived from environment variables/managed identity via DefaultAzureCredential.\n    connection_string = os.environ.get('AZURE_STORAGE_CONNECTION_STRING', '')\n    \n    if not connection_string:\n        print(\"Warning: AZURE_STORAGE_CONNECTION_STRING environment variable not set. Using dummy string.\")\n        # Fallback for example, in real world this would likely fail or use DefaultAzureCredential\n        blob_storage_credentials = AzureBlobStorageCredentials()\n    else:\n        blob_storage_credentials = AzureBlobStorageCredentials(connection_string=connection_string)\n    \n    # If the AzureBlobStorageContainer block was saved via the UI or another script:\n    # my_blob_container_block = AzureBlobStorageContainer.load(\"my-blob-container\")\n    # Instead, we create an ad-hoc one for this example's simplicity\n    my_blob_container_block = AzureBlobStorageContainer(\n        container_name=\"prefect\", # Replace with your container name\n        credentials=blob_storage_credentials\n    )\n\n    print(f\"Attempting to download 'prefect.txt' from container 'prefect'...\")\n    data = blob_storage_download(\n        blob=\"prefect.txt\", # Replace with your blob name\n        container=\"prefect\", # Or use my_blob_container_block directly if it defines the container\n        blob_storage_credentials=blob_storage_credentials,\n    )\n    print(f\"Downloaded data (first 100 chars): {data[:100].decode()}\")\n    return data\n\nif __name__ == \"__main__\":\n    # Set a dummy connection string for local testing if not already set\n    # For actual Azure access, replace with a real connection string or use Azure identity management\n    # os.environ['AZURE_STORAGE_CONNECTION_STRING'] = 'DefaultEndpointsProtocol=https;AccountName=youraccount;AccountKey=yourkey;EndpointSuffix=core.windows.net'\n    example_blob_storage_download_flow()","lang":"python","description":"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."},"warnings":[{"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.","message":"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.","severity":"breaking","affected_versions":"<2.0.0 (Prefect core)"},{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions of `prefect-azure` with Prefect 2.x+"},{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"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\")`.","message":"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.","severity":"gotcha","affected_versions":"All versions of `prefect-azure` with Prefect 2.x+"}],"env_vars":null,"last_verified":"2026-04-15T00:00:00.000Z","next_check":"2026-07-14T00:00:00.000Z"}