feat(data-factory): update connection handling to support item-specific connections and enhance documentation

Author: aviraj-gour

metadata-ingestion/docs/sources/fabric-data-factory/fabric-data-factory_pre.md

@@ -6,7 +6,7 @@ The `fabric-data-factory` module ingests metadata from Microsoft Fabric Data Fac
 
 1. **Set up authentication** — Configure Azure credentials (see [Prerequisites](#prerequisites))
 2. **Enable API access** — Ensure a Fabric admin has enabled service principal API access (if using SP or managed identity)
-3. **Grant permissions** — Add your identity as a workspace member with Viewer role or higher
+3. **Grant permissions** — Add your identity as a workspace **Contributor** (required for pipeline definitions and lineage)
 4. **Configure recipe** — Use `fabric-data-factory_recipe.yml` as a template
 5. **Run ingestion** — Execute `datahub ingest -c fabric-data-factory_recipe.yml`
    :::
@@ -41,33 +41,29 @@ Fabric Data Factory Concepts
 
 #### Required Permissions
 
-The connector requires read access to Fabric workspaces, items, and connections. The specific permissions depend on your authentication method, but the identity you use must have:
-
-- **Workspace access**: The identity must be added as a workspace member (Viewer or above) for each workspace you want to ingest
+The connector requires **Contributor** role on each workspace. Contributor is needed to fetch pipeline definitions without it, the connector will list pipelines but fail to read their activities and lineage.
 
 ##### Delegated (on behalf of a user) authentication
 
-If using delegated auth (e.g., Azure CLI), request the following Fabric API scopes via Microsoft Entra ID:
+If using delegated auth (e.g., Azure CLI), the signed-in user's existing Fabric permissions apply directly. The connector requires the following delegated scopes:
 
-- `Workspace.Read.All` — to list and read workspaces
-- `Item.Read.All` — to list and read items (pipelines, activities)
-- `Connection.Read.All` — to list and read connections for lineage resolution
-  For execution history, additionally request:
+- `Workspace.Read.All` or `Workspace.ReadWrite.All` — for listing workspaces and items
+- `Item.ReadWrite.All` or `DataPipeline.ReadWrite.All` — for Get Item Definition, List Item Connections, and Query Activity Runs (`Item.Read.All` is **not** sufficient for definitions and connections)
+- `Item.Read.All` or `DataPipeline.Read.All` — sufficient for List Item Job Instances (execution history)
 
-- `Item.Execute.All` — to query pipeline and activity runs
+The Azure CLI token includes the necessary Fabric API scopes by default.
 
 ##### Service Principal and Managed Identity authentication
 
-Service principals and managed identities do not use delegated scopes. Instead, you need to:
+Service principals and managed identities do not inherit any permissions by default. You need to:
 
 1. **Enable API access**: A Fabric admin must enable the service principal tenant settings (see **Fabric Admin Settings** below)
-2. **Grant workspace access**: Add the SP or MI as a workspace member (Viewer or above) for each workspace you want to ingest
-3. **Grant connection access**: The SP or MI must have permission on the Fabric connections used by pipelines, so that the connector can read connection details for lineage resolution
+2. **Grant workspace access**: Add the SP or MI as a workspace **Contributor** for each workspace you want to ingest
 
 #### Fabric Admin Settings
 
 :::warning
-For **service principal** and **managed identity** authentication, a Fabric administrator must enable API access for service principals in the Fabric admin portal. Without this, API calls will fail with 401/403 errors even if permissions are correctly assigned.
+For **service principal** and **managed identity** authentication, a Fabric administrator must enable API access for service principals in the Fabric admin portal. Without this, API calls will fail with 401 errors even if workspace permissions are correctly assigned.
 :::
 
 As of mid-2025, Microsoft split the original single tenant setting into two separate settings. Configure them as follows:
@@ -76,10 +72,12 @@ As of mid-2025, Microsoft split the original single tenant setting into two sepa
 2. Under **Developer settings**, enable the applicable setting(s):
    - **Service principals can call Fabric public APIs** — Controls access to CRUD APIs protected by the Fabric permission model (e.g., reading workspaces and items). This is **enabled by default** for new tenants since August 2025.
    - **Service principals can create workspaces, connections, and deployment pipelines** — Controls access to global APIs not protected by Fabric permissions. This is **disabled by default**. Enable only if needed.
-3. Choose whether to enable for the entire organization or restrict to a specific security group. It is recommended to restrict access to a dedicated security group containing only the service principals that need API access.
+3. Restrict access to a dedicated **security group** containing only the service principals that need API access. This is the recommended approach.
 
 > **Note**: If you are on an older tenant where the legacy single setting **Service principals can use Fabric APIs** is still visible, enable that instead. It will be automatically migrated to the two new settings.
 
+> **Note**: Tenant setting changes can take **up to 15 minutes** to propagate. If you receive 401 errors immediately after enabling, wait and retry.
+
 For detailed instructions, see [Developer admin settings](https://learn.microsoft.com/en-us/fabric/admin/service-admin-portal-developer) and [Identity support for Fabric REST APIs](https://learn.microsoft.com/en-us/rest/api/fabric/articles/identity-support).
 
 #### Authentication
@@ -88,7 +86,11 @@ The connector supports four authentication methods via the shared `credential` c
 
 ##### Service Principal (recommended for production)
 
-Register an application in [Microsoft Entra ID](https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app) and note the `client_id`, `client_secret`, and `tenant_id`. Then ensure the Fabric admin has enabled service principal API access (see **Fabric Admin Settings** above) and add the service principal as a member of the target workspaces.
+Register an application in [Microsoft Entra ID](https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app) and note the `client_id`, `client_secret`, and `tenant_id`. Then:
+
+1. Ensure the Fabric admin has enabled service principal API access (see **Fabric Admin Settings** above)
+2. Create a security group in Entra ID and add the service principal as a member
+3. Add the security group as **Contributor** in each target workspace (Contributor role grants access to pipeline definitions and item connections for lineage)
 
 ```yaml
 credential:
@@ -102,7 +104,7 @@ All three fields are required when using this method.
 
 ##### Managed Identity (for Azure-hosted deployments)
 
-Use this when running DataHub ingestion on an Azure VM, AKS, App Service, or other Azure compute that supports [managed identities](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/overview). The managed identity must be added as a workspace member in Fabric. A Fabric admin must also enable the tenant settings described in **Fabric Admin Settings** above — these settings govern API access for both service principals and managed identities, despite the setting name referencing only service principals.
+Use this when running DataHub ingestion on an Azure VM, AKS, App Service, or other Azure compute that supports [managed identities](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/overview). The managed identity must be added as a workspace Contributor in Fabric. A Fabric admin must also enable the tenant settings described in **Fabric Admin Settings** above — these settings govern API access for both service principals and managed identities, despite the setting name referencing only service principals.
 
 ```yaml
 # System-assigned managed identity (no additional config needed)
@@ -118,15 +120,17 @@ credential:
   managed_identity_client_id: "<your-managed-identity-client-id>"
 ```
 
-##### Azure CLI (for local development)
+##### Azure CLI (for local development and testing)
 
-Uses the credentials from your local `az login` session. Run `az login` before starting ingestion. The signed-in user must have workspace access in Fabric.
+Uses the credentials from your local `az login` session. The signed-in user's existing Fabric permissions apply directly — no additional setup needed beyond workspace access.
 
 ```yaml
 credential:
   authentication_method: cli
 ```
 
+Run `az login` before starting ingestion. For remote servers without a browser, use `az login`.
+
 ##### DefaultAzureCredential (flexible auto-detection)
 
 Uses Azure's [DefaultAzureCredential](https://learn.microsoft.com/en-us/python/api/azure-identity/azure.identity.defaultazurecredential) chain, which tries multiple credential sources in order: environment variables, workload identity, managed identity, shared token cache, Azure CLI, Azure PowerShell, Azure Developer CLI, and more.
@@ -149,6 +153,8 @@ credential:
 #### Setup
 
 1. Choose an authentication method from above and configure the `credential` block.
-2. If using service principal or managed identity, ensure the Fabric admin has enabled the appropriate developer settings for service principal API access (see **Fabric Admin Settings** above).
-3. Add the identity as a member of the target workspaces with Viewer role or above.
+2. If using service principal or managed identity:
+   - Ensure the Fabric admin has enabled the appropriate developer settings (see **Fabric Admin Settings**)
+   - Create a security group, add your identity, and grant **Contributor** on target workspaces
+3. If using Azure CLI, run `az login` (or `az login --use-device-code` on remote servers).
 4. Configure the ingestion recipe with optional workspace and pipeline filters.

metadata-ingestion/src/datahub/ingestion/source/fabric/common/core_client.py

@@ -132,24 +132,37 @@ def get_item_definition(
 
         return result
 
-    def list_connections(self) -> Iterator[FabricConnection]:
-        """List all tenant-scoped connections accessible to the caller.
+    def list_item_connections(
+        self, workspace_id: str, item_id: str
+    ) -> Iterator[FabricConnection]:
+        """List connections for a specific item.
+
+        Returns only connections bound to the given item.
 
-        Connections are referenced by pipeline activities via
-        externalReferences.connection (the connection GUID).
+        Some connections (e.g. Automatic/SSO) may not have an id —
+        these are skipped since they cannot be referenced by activities.
 
-        Reference: https://learn.microsoft.com/en-us/rest/api/fabric/core/connections/list-connections
+        Reference: https://learn.microsoft.com/en-us/rest/api/fabric/core/items/list-item-connections
+
+        Args:
+            workspace_id: Workspace GUID
+            item_id: Item GUID
 
         Yields:
-            FabricConnection objects
+            FabricConnection objects (only those with an id)
         """
-        logger.info("Listing Fabric connections")
-        for raw in self._paginate("connections"):
+        endpoint = f"workspaces/{workspace_id}/items/{item_id}/connections"
+        logger.debug(f"Listing connections for item {item_id}")
+        for raw in self._paginate(endpoint):
+            # Skip connections without an id (e.g. Automatic/SSO)
+            if not raw.get("id"):
+                continue
             try:
                 yield FabricConnection.from_dict(raw)
             except KeyError as e:
                 self.report.report_parse_failure(
-                    f"Skipping malformed connection: missing required field {e}"
+                    f"Skipping malformed item connection for "
+                    f"item {item_id}: missing required field {e}"
                 )
 
     def list_item_job_instances(

metadata-ingestion/src/datahub/ingestion/source/fabric/data_factory/source.py

@@ -190,8 +190,9 @@ def __init__(self, config: FabricDataFactorySourceConfig, ctx: PipelineContext):
         self.report.client_report = self.client_report
 
         # Connection cache: connection GUID → FabricConnection
-        # Populated once before processing pipelines, used by lineage
-        # resolvers to map activity connection refs to DataHub platforms.
+        # Populated per-item via list_item_connections during pipeline
+        # fetching, used by lineage resolvers to map activity connection
+        # refs to DataHub platforms.
         self._connections_cache: dict[str, FabricConnection] = {}
 
         # Pipeline run ID → DPI URN, populated during pipeline run emission,
@@ -209,21 +210,6 @@ def __init__(self, config: FabricDataFactorySourceConfig, ctx: PipelineContext):
         # when emitting the child activity's DataJobInputOutput to merge the edge in.
         self._cross_pipeline_edges: dict[str, list[str]] = {}
 
-    def _cache_connections(self) -> None:
-        """Fetch all tenant connections and cache them keyed by ID."""
-        try:
-            for conn in self.client.list_connections():
-                self._connections_cache[conn.id] = conn
-            logger.info(f"Cached {len(self._connections_cache)} Fabric connection(s)")
-        except Exception as e:
-            self.report.report_warning(
-                title="Failed to Cache Connections",
-                message="Could not retrieve tenant connections. "
-                "Copy activity lineage may be incomplete.",
-                context="",
-                exc=e,
-            )
-
     @classmethod
     def create(
         cls, config_dict: dict, ctx: PipelineContext
@@ -251,9 +237,8 @@ def get_workunits_internal(
         logger.info("Starting Fabric Data Factory ingestion")
 
         try:
-            # Lineage extractors and connection cache are only needed when lineage is enabled
+            # Lineage extractors are only needed when lineage is enabled
             if self.config.include_lineage:
-                self._cache_connections()
                 self._copy_lineage_extractor = CopyActivityLineageExtractor(
                     connections_cache=self._connections_cache,
                     report=self.report,
@@ -324,7 +309,8 @@ def _fetch_pipeline_activities(self, workspace_id: str) -> list[FabricItem]:
         """Fetch pipelines and their activities for a workspace into cache.
 
         Applies pipeline_pattern filter. Populates
-        ``self._pipeline_activities_cache``.
+        ``self._pipeline_activities_cache``. When lineage is enabled,
+        also fetches per-item connections into ``self._connections_cache``.
         """
         pipeline_items: list[FabricItem] = []
         for item in self.client.list_items(
@@ -349,8 +335,26 @@ def _fetch_pipeline_activities(self, workspace_id: str) -> list[FabricItem]:
                     exc=e,
                 )
                 self._pipeline_activities_cache[(workspace_id, item.id)] = []
+
+            if self.config.include_lineage:
+                self._cache_item_connections(workspace_id, item)
         return pipeline_items
 
+    def _cache_item_connections(self, workspace_id: str, item: FabricItem) -> None:
+        """Fetch per-item connections and add to the connections cache."""
+        try:
+            for conn in self.client.list_item_connections(workspace_id, item.id):
+                if conn.id not in self._connections_cache:
+                    self._connections_cache[conn.id] = conn
+        except Exception as e:
+            self.report.report_warning(
+                title="Failed to Fetch Item Connections",
+                message="Could not retrieve connections for item. "
+                "Lineage may be incomplete for this pipeline.",
+                context=f"pipeline={item.name}",
+                exc=e,
+            )
+
     def _create_workspace_container(
         self, workspace: FabricWorkspace
     ) -> Iterable[Entity]:

metadata-ingestion/tests/integration/fabric_data_factory/test_fabric_data_factory_source.py

@@ -201,7 +201,7 @@ def activities_side_effect(workspace_id: str, pipeline_id: str) -> List[Any]:
         ),
         patch.object(
             FabricDataFactoryClient,
-            "list_connections",
+            "list_item_connections",
             return_value=iter(connections),
         ),
         patch.object(
@@ -266,7 +266,7 @@ def test_no_execution_history(pytestconfig: pytest.Config, tmp_path: Path) -> No
         ),
         patch.object(
             FabricDataFactoryClient,
-            "list_connections",
+            "list_item_connections",
             return_value=iter([]),
         ),
         patch.object(

metadata-ingestion/tests/integration/fabric_data_factory/test_factories.py

@@ -56,7 +56,7 @@ def create_connection(
     connection_type: str,
     path: Optional[str] = None,
 ) -> FabricConnection:
-    """Create a FabricConnection matching list_connections() output."""
+    """Create a FabricConnection matching list_item_connections() output."""
     return FabricConnection(
         id=connection_id,
         display_name=display_name,