Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
# Copyright 2025 Collate
# Licensed under the Collate Community License, Version 1.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Dashboard service step identity and shared check helpers.

Dashboard connectors reach their source over a REST API (with OAuth or token
auth), not a SQLAlchemy engine, so these helpers stay HTTP-generic: they call a
connector-supplied callable and summarize what it returned, and on failure raise
``CheckError`` carrying the label of the call they attempted, so a failed step
still reports its ``Evidence`` to the backend. The runner and the rest of the
package stay engine- and protocol-agnostic; connectors reuse these helpers from
their ``@check`` methods.
"""

from __future__ import annotations

from typing import TYPE_CHECKING

from metadata.core.connections.test_connection.check import CheckError, StepName
from metadata.core.connections.test_connection.records import Diagnosis, Evidence

if TYPE_CHECKING:
from collections.abc import Callable, Sized


# A check only needs to prove the list endpoint is reachable and returns items,
# not enumerate every one, so ``fetch_list`` counts at most this many and renders
# ``<cap>+`` when the count meets or exceeds it, keeping the summary bounded on
# huge tenants.
DEFAULT_LIST_CAP = 100


class DashboardStep(StepName):
"""The steps a dashboard connector can be asked to verify."""

CheckAccess = "CheckAccess"
GetDashboards = "GetDashboards"


def _count(number: int, noun: str, cap: int | None = None) -> str:
"""``3 dashboards`` / ``1 dashboard`` - pluralize the noun to match the count.

When ``cap`` is given and the count meets or exceeds it, the figure is
rendered ``<cap>+`` so a capped sample is not read as an exact total. The
caller caps ``number`` at ``cap`` first, so this only ever renders ``<cap>+``
at the boundary, never a larger bare number.
"""
plural = noun if number == 1 else noun + "s"
shown = f"{cap}+" if cap is not None and number >= cap else str(number)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Small thing: with exactly 100 dashboards this shows "100+". Since the count is exact, > instead of >= keeps it a plain "100".

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The idea here is to cap the return at 100, so > would never be true

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in 777f978 along @IceS2's line — the counted value is now capped (min(len, cap)), so it can never exceed the cap and > would never fire; >= renders the exact 100+ boundary. Docstrings updated to "meets or exceeds".

return f"{shown} {plural}"


def verify_access(authenticate: Callable[[], object], command: str) -> Evidence:
"""Prove the connector can authenticate against the REST API.

Runs the connector's auth callable (e.g. an OAuth token acquisition) and, on
failure, re-raises as ``CheckError`` carrying the attempted command so the
gate step still reports what it tried.
"""
try:
authenticate()
except Exception as cause:
raise CheckError(cause, Evidence(command=command)) from cause
return Evidence(summary="authenticated", command=command)


def fetch_list(
fetch: Callable[[], Sized | None],
noun: str,
command: str,
cap: int = DEFAULT_LIST_CAP,
empty_caveat: Diagnosis | None = None,
) -> Evidence:
"""Call a REST list endpoint and report how many items it returned.

Reports the command it attempted and a count summary; the count is capped at
``cap`` and rendered ``<cap>+`` when it meets or exceeds it, so a huge tenant
does not produce an unbounded figure. When the endpoint returns nothing and
``empty_caveat`` is given, the step still passes but carries the caveat as a
non-blocking Warning - a connector opts in when "nothing visible" is worth
surfacing (e.g. it may mean the fetch was filtered or silently failed rather
than the source being genuinely empty). On failure, re-raise as ``CheckError``
carrying the command so the failed step still reports what it ran.
"""
try:
items = fetch()
except Exception as cause:
raise CheckError(cause, Evidence(command=command)) from cause
count = min(len(items) if items else 0, cap)
caveat = empty_caveat if count == 0 else None
return Evidence(summary=f"{_count(count, noun, cap)} enumerated", command=command, caveat=caveat)
Original file line number Diff line number Diff line change
Expand Up @@ -13,26 +13,120 @@
Source connection handler
"""

from typing import Optional
from __future__ import annotations

from metadata.generated.schema.entity.automations.workflow import (
Workflow as AutomationWorkflow,
from typing import TYPE_CHECKING

from metadata.core.connections.test_connection import (
Diagnosis,
ErrorPack,
Evidence,
Matchers,
check,
when,
)
from metadata.core.connections.test_connection.checks.dashboard import (
DashboardStep,
fetch_list,
verify_access,
)
from metadata.core.connections.test_connection.classifier import exception_chain
from metadata.core.connections.test_connection.network import NETWORK_ERRORS
from metadata.generated.schema.entity.services.connections.dashboard.powerBIConnection import (
PowerBIConnection as PowerBIConnectionConfig,
)
from metadata.generated.schema.entity.services.connections.testConnectionResult import (
TestConnectionResult,
)
from metadata.ingestion.api.steps import InvalidSourceException
from metadata.ingestion.connections.connection import BaseConnection
from metadata.ingestion.connections.test_connections import test_connection_steps
from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.ingestion.source.dashboard.powerbi.client import (
PowerBiApiClient,
PowerBiClient,
)
from metadata.ingestion.source.dashboard.powerbi.file_client import PowerBiFileClient
from metadata.utils.constants import THREE_MIN

if TYPE_CHECKING:
from metadata.core.connections.test_connection import ChecksProvider
from metadata.core.connections.test_connection.classifier import Matcher


def _http_status(*codes: int) -> Matcher:
"""Match a PowerBI REST error by HTTP status.

The status is the stable signal - the message body varies by tenant and
endpoint. Two shapes carry it: the client's ``APIError`` exposes it on a
top-level ``.status_code`` property, but when the error body has no ``code``
field the client re-raises the raw ``requests.HTTPError`` unchanged, which
carries the status at ``.response.status_code``. We consult both, across the
cause chain."""
wanted = frozenset(codes)

def match(error: BaseException) -> bool:
for current in exception_chain(error):
code = getattr(current, "status_code", None)
if code is None:
response = getattr(current, "response", None)
code = getattr(response, "status_code", None)
if isinstance(code, int) and code in wanted:
return True
return False

return match


def _contains_any(*tokens: str) -> Matcher:
"""Match when any of ``tokens`` appears in the error (or its cause chain).

MSAL reports a bad authority through more than one message shape - a malformed
tenant, a blank tenant - so a single ``Matchers.contains`` token would miss
some; this ORs the stable tokens across those shapes."""
lowered = tuple(token.lower() for token in tokens)

def match(error: BaseException) -> bool:
chain = " ".join(str(current) for current in exception_chain(error)).lower()
return any(token in chain for token in lowered)

return match


POWERBI_ERRORS = ErrorPack(
# CheckAccess acquires the OAuth token, so a bad secret fails there as an
# InvalidSourceException - not here. Any 401/403 from a REST call therefore
# means the token was accepted but Power BI would not authorize the call, so
# neither diagnosis points at the credentials. Per Microsoft's REST API
# troubleshooting guide, a service principal that is not enabled for the
# (admin) APIs returns 401, while a missing workspace/resource grant returns
# 403 - so the two split on tenant-enablement vs. workspace access.
when(Matchers.exception(InvalidSourceException)).diagnose(
"Authentication failed",
fix="Could not acquire an OAuth token. Check the Client ID, Client Secret, and Tenant ID, "
"and that the app registration is allowed to request the configured scope.",
Comment thread
gitar-bot[bot] marked this conversation as resolved.
),
when(_http_status(401)).diagnose(
"Power BI did not authorize the service principal",
fix="The token was accepted but Power BI rejected the call (401). In the Fabric admin "
"portal enable 'Allow service principals to use Power BI APIs' (and the read-only admin "
"APIs setting when Use Admin APIs is on), add the service principal to the allowed "
"security group, and grant the app registration the required API permission.",
),
when(_http_status(403)).diagnose(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does a service principal with no access actually return 403 here, or 401? MS APIs often give 401. Worth confirming against a real tenant before merge — the tests can't tell us which one PowerBI really sends.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confirmed against the MS docs and fixed in 5456918. Microsoft's Troubleshoot Power BI REST APIs documents that a service principal not enabled for the (admin) APIs returns 401 (a tenant/app-registration gap, not an auth failure), while a missing workspace grant returns 403. Since CheckAccess acquires the token first, any 401/403 here is post-auth, so neither points at the secret: 401 → "Power BI did not authorize the service principal", 403 → "Insufficient permissions". Still worth a live confirm of the exact code per endpoint.

"Insufficient permissions",
fix="The service principal is authenticated but not authorized for this resource (403). "
"Add it as a member (or admin) of the target workspace and grant the permissions the "
"call needs.",
),
when(_http_status(404)).diagnose(
"Resource not found",
fix="The requested resource was not found (404). Check the API URL and that the configured "
"tenant/workspace exists and is visible to the service principal.",
),
# Kept last: authority/instance-discovery failures are MSAL ValueErrors that
# carry no HTTP status, so a broad substring match here must not shadow a
# status-coded 401/403/404 whose message happens to echo the authority URL.
when(_contains_any("authority", "should consist of an https url")).diagnose(
"Invalid tenant or authority",
fix="The authority could not be resolved. Check the Tenant ID (a valid GUID or "
"tenant name) and the Authority URI, e.g. https://login.microsoftonline.com/.",
),
).including(NETWORK_ERRORS)


def get_connection(connection: PowerBIConnectionConfig) -> PowerBiClient:
Expand All @@ -45,28 +139,61 @@ def get_connection(connection: PowerBIConnectionConfig) -> PowerBiClient:
return PowerBiClient(api_client=PowerBiApiClient(connection), file_client=file_client)


class PowerBIChecks:
"""Test-connection checks for PowerBI.

``CheckAccess`` is the gate: it acquires an OAuth token, so a bad service
principal fails fast before any list endpoint is dialled. ``GetDashboards``
then exercises list access.

The REST client is built lazily inside the first check, never at
construction: the underlying MSAL client performs authority/instance
discovery over the network in its constructor, so building it eagerly would
run before the runner's gate and surface as a raw workflow error instead of a
classified ``CheckAccess`` failure.
"""

errors = POWERBI_ERRORS

def __init__(self, connection: PowerBIConnectionConfig) -> None:
self._connection = connection
self._api_client: PowerBiApiClient | None = None

def _client(self) -> PowerBiApiClient:
"""Build (once) and return the REST client. Built directly rather than via
``get_connection`` so the file client (unused by the checks) is never
constructed. The MSAL client it wraps does authority/instance discovery
over the network in its constructor, so this is only ever called from
inside a check - never at construction."""
if self._api_client is None:
self._api_client = PowerBiApiClient(self._connection)
return self._api_client
Comment thread
greptile-apps[bot] marked this conversation as resolved.

@check(DashboardStep.CheckAccess)
def check_access(self) -> Evidence:
return verify_access(
lambda: self._client().get_auth_token(),
command="acquire OAuth token",
)

@check(DashboardStep.GetDashboards)
def get_dashboards(self) -> Evidence:
return fetch_list(
self._client().fetch_dashboards,
noun="dashboard",
command="fetch dashboards",
empty_caveat=Diagnosis(
title="No dashboards visible",
remediation="The connection works but returned no dashboards. Confirm the service "
"principal has access to a workspace that contains dashboards - an empty result can "
"also mean the listing was filtered or could not be read.",
),
)

Comment thread
IceS2 marked this conversation as resolved.

class PowerBIConnection(BaseConnection[PowerBIConnectionConfig, PowerBiClient]):
def _get_client(self) -> PowerBiClient:
return get_connection(self.service_connection)

def test_connection(
self,
metadata: OpenMetadata,
automation_workflow: Optional[AutomationWorkflow] = None, # noqa: UP045
timeout_seconds: Optional[int] = THREE_MIN, # noqa: UP045
) -> TestConnectionResult:
"""
Test connection. This can be executed either as part
of a metadata workflow or during an Automation Workflow
"""
client = self.client
service_connection = self.service_connection
test_fn = {"GetDashboards": client.api_client.fetch_dashboards}

return test_connection_steps(
metadata=metadata,
test_fn=test_fn,
service_type=service_connection.type.value, # pyright: ignore[reportOptionalMemberAccess]
automation_workflow=automation_workflow,
timeout_seconds=timeout_seconds,
)
def checks(self) -> ChecksProvider:
return PowerBIChecks(connection=self.service_connection)
Loading
Loading