"""OWS Permissions connector for checking profile access.""" from typing import Any from uuid import UUID from contributor.api import context, datasources ALL_ACCESS_VENDOR_ID = "*" ALL_ACCESS_VENDOR_UUID = "053a1a75-acc5-4cd8-9206-a194335d2afa" def get_directly_accessible_vendors() -> dict[str, Any]: """ Get vendors the authenticated profile has direct access to. """ request_ctx = context.get_request_context() headers = {} if request_ctx: if request_ctx.profile_type: headers["Orchard-Profile-Type"] = request_ctx.profile_type if request_ctx.profile_id: headers["Orchard-Profile-Id"] = str(request_ctx.profile_id) if request_ctx.identity_id: headers["Orchard-Identity-Id"] = request_ctx.identity_id if request_ctx.identity_uuid: headers["Orchard-Identity-Uuid"] = request_ctx.identity_uuid result = datasources.get_ows_client().get( "ows-permissions", path="/v2/profile/self/vendors/direct-access", headers=headers, ) result.raise_for_status() return result.json() def can_access_vendor( *, vendor_id: int | None = None, vendor_uuid: UUID | None = None ) -> bool: """ Check if authenticated profile can access a specific vendor. Args: vendor_id: Numeric vendor ID to check access for. vendor_uuid: UUID vendor identifier to check access for. Returns: True when a matching vendor (or all-access vendor) is found, else False. Note: Currently delegates to get_directly_accessible_vendors() and filters client-side. This should eventually be replaced by some kind of permissions call TBD. """ accessible = get_directly_accessible_vendors() vendors = accessible.get("vendors", []) if vendor_id: match = next( ( v for v in vendors if str(v.get("vendor_id")) in {str(vendor_id), ALL_ACCESS_VENDOR_ID} ), None, ) print("vendor_id", match) return match is not None if vendor_uuid: vendor_uuid_str = str(vendor_uuid) match = next( ( v for v in vendors if str(v.get("vendor_uuid")) in {vendor_uuid_str, ALL_ACCESS_VENDOR_UUID} ), None, ) print("vendor_uuid", match) return match is not None return False