Azure Data Factory: a copy pipeline in Python

I built a small Azure Data Factory (ADF) pipeline entirely from Python: create the factory, connect it to Blob Storage, define input and output datasets, run a copy activity, then poll the run until it finishes. This page explains the design and the code, and what I'd change before running it in production.

Architecture

The code

This version uses azure-identity. DefaultAzureCredential picks up a managed identity when the code runs in Azure, and my az login session on a laptop, so no secret is ever written in code or typed into a form.

from azure.identity import DefaultAzureCredential
from azure.mgmt.datafactory import DataFactoryManagementClient
from azure.mgmt.datafactory.models import (
    AzureBlobDataset, AzureBlobStorageLinkedService, BlobSink, BlobSource, CopyActivity,
    DatasetReference, DatasetResource, LinkedServiceReference, LinkedServiceResource,
    PipelineResource)

credential = DefaultAzureCredential()        # managed identity in Azure, `az login` locally
adf = DataFactoryManagementClient(credential, SUBSCRIPTION_ID)

# Linked service: with only an endpoint, ADF authenticates with the factory's own
# managed identity (granted "Storage Blob Data Contributor"), so no account key exists.
storage = AzureBlobStorageLinkedService(service_endpoint=f"https://{ACCOUNT}.blob.core.windows.net")
adf.linked_services.create_or_update(RG, FACTORY, "blobStorage", LinkedServiceResource(properties=storage))

ls = LinkedServiceReference(type="LinkedServiceReference", reference_name="blobStorage")
for name, folder in [("ds_in", "adftutorial/input"), ("ds_out", "adftutorial/output")]:
    dataset = AzureBlobDataset(linked_service_name=ls, folder_path=folder)
    adf.datasets.create_or_update(RG, FACTORY, name, DatasetResource(properties=dataset))

copy = CopyActivity(
    name="copyBlobToBlob",
    inputs=[DatasetReference(type="DatasetReference", reference_name="ds_in")],
    outputs=[DatasetReference(type="DatasetReference", reference_name="ds_out")],
    source=BlobSource(), sink=BlobSink())
adf.pipelines.create_or_update(RG, FACTORY, "copyPipeline", PipelineResource(activities=[copy]))

run = adf.pipelines.create_run(RG, FACTORY, "copyPipeline", parameters={})
status = adf.pipeline_runs.get(RG, FACTORY, run.run_id).status   # Queued → InProgress → Succeeded

My first version (2020) used a service-principal secret and a storage connection string. Moving to DefaultAzureCredential and managed identity removes both.

What I'd do in production

Area	Choice
Identity	Managed identities only. Least-privilege role assignments scoped to the one container, with no keys or SAS tokens.
Infrastructure	The factory, storage, role assignments and pipeline defined in Bicep or Terraform, deployed from CI with a federated (OIDC) credential.
Pipeline code	ADF Git integration. Changes are reviewed in pull requests and promoted from dev to test to prod through ARM template parameters.
Monitoring	Diagnostic settings to Log Analytics, alerts on failed runs and on long durations, and a run-history dashboard.
Data	Parameterised paths and incremental copies (watermark or file-date folders) instead of whole-folder copies, and a dead-letter folder for bad files.
Network	Private endpoints for storage, and a managed virtual network for the integration runtime.

An earlier version of this page let visitors run the pipeline in their own subscription by entering credentials. I removed it: a portfolio site should never ask for anyone's cloud secrets.