yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-11-01 19:18:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

powerbi change owner condition (#21724)

Browse Source
This commit is contained in:
harshsoni2024 2025-06-12 16:11:43 +05:30 committed by GitHub
parent 62d91fbf2e
commit 6a6180b2e3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 63 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
ingestion/src/metadata/ingestion/source/dashboard/powerbi/metadata.py
View File

@ -79,6 +79,8 @@ from metadata.utils.logger import ingestion_logger
logger = ingestion_logger()
OWNER_ACCESS_RIGHTS_KEYWORDS = ["owner", "write", "admin"]
class PowerbiSource(DashboardServiceSource):
"""PowerBi Source Class"""
@ -1010,42 +1012,49 @@ class PowerbiSource(DashboardServiceSource):
elif isinstance(dashboard_details, PowerBIDashboard):
access_right = owner.dashboardUserAccessRight
if owner.userType != "Member" or (
isinstance(
dashboard_details, (Dataflow, PowerBIReport, PowerBIDashboard)
)
and access_right != "Owner"
):
if owner.userType != "Member":
logger.warning(
f"User is not a member and has no access to the {dashboard_details.id}: ({owner.displayName}, {owner.email})"
f"User is not a member of {dashboard_details.id}:"
f" ({owner.displayName}, {owner.email})"
)
continue
if owner.email:
try:
owner_email = EmailStr._validate(owner.email)
except PydanticCustomError:
logger.warning(f"Invalid email for owner: {owner.email}")
owner_email = None
if owner_email:
if access_right and any(
keyword in access_right.lower()
for keyword in OWNER_ACCESS_RIGHTS_KEYWORDS
):
if owner.email:
try:
owner_ref = self.metadata.get_reference_by_email(
owner_email.lower()
owner_email = EmailStr._validate(owner.email)
except PydanticCustomError:
logger.warning(f"Invalid email for owner: {owner.email}")
owner_email = None
if owner_email:
try:
owner_ref = self.metadata.get_reference_by_email(
owner_email.lower()
)
except Exception as err:
logger.warning(
f"Could not process owner data with email"
f" {owner.email} in {dashboard_details.id}: {err}"
)
elif owner.displayName:
try:
owner_ref = self.metadata.get_reference_by_name(
name=owner.displayName
)
except Exception as err:
logger.warning(
f"Could not fetch owner data with email {owner.email} in {dashboard_details.id}: {err}"
f"Could not process owner data with name"
f" {owner.displayName} in {dashboard_details.id}: {err}"
)
elif owner.displayName:
try:
owner_ref = self.metadata.get_reference_by_name(
name=owner.displayName
)
except Exception as err:
logger.warning(
f"Could not process owner data with name {owner.displayName} in {dashboard_details.id}: {err}"
)
if owner_ref:
owner_ref_list.append(owner_ref.root[0])
if owner_ref:
owner_ref_list.append(owner_ref.root[0])
else:
logger.warning(
f"User does not have owner, admin or write access to"
f" {dashboard_details.id}: ({owner.displayName}, {owner.email})"
)
# check for last modified, configuredBy user
current_active_user = None
if isinstance(dashboard_details, Dataset):

26
ingestion/tests/unit/topology/dashboard/test_powerbi.py
View File

@ -311,6 +311,32 @@ class PowerBIUnitTest(TestCase):
# Verify get_reference_by_email was not called when there are no owners
self.powerbi.metadata.get_reference_by_email.assert_not_called()
# Reset mock for invalid owners test
self.powerbi.metadata.get_reference_by_email.reset_mock()
# Test with invalid owners
dashboard_invalid_owners = PowerBIDashboard.model_validate(
{
"id": "dashboard3",
"displayName": "Test Dashboard 3",
"webUrl": "https://test.com",
"embedUrl": "https://test.com/embed",
"tiles": [],
"users": [
{
"displayName": "Kane Williams",
"emailAddress": "kane.williams@example.com",
"dashboardUserAccessRight": "Read",
"userType": "Member",
},
],
}
)
owner_ref = self.powerbi.get_owner_ref(dashboard_invalid_owners)
self.assertIsNone(owner_ref)
# Verify get_reference_by_email was not called when there are no owners
self.powerbi.metadata.get_reference_by_email.assert_not_called()
@pytest.mark.order(3)
def test_parse_table_info_from_source_exp(self):
table = PowerBiTable(