from fansifter_common.auth.account import Account, AccountAccess from fansifter_common.auth.authorization import AuthorizationBackend from fansifter_common.auth.exceptions import PermissionDenied from fansifter_common.auth.types import JointVentureParticipant, Permission class AuthService: def __init__(self, authorization_backend: AuthorizationBackend) -> None: self.authorization_backend = authorization_backend def authorize_for_permission( self, identity_id: str, /, *, permission: Permission ) -> AccountAccess: return self.authorization_backend.authorize_for_resource_type_action( identity_id, resource_type=permission.resource_type, action=permission.action, ) def authorize_account( self, identity_id: str, /, *, permission: Permission, account: Account ) -> AccountAccess: """Authorize a user for specific account with resource type actions and return the account access.""" return self.authorization_backend.authorize_account( identity_id, resource_type=permission.resource_type, action=permission.action, account=account, ) def authorize_optional_account( self, identity_id: str, /, *, permission: Permission, account: Account | None ) -> AccountAccess: """Authorize a user for optional account and return the account access.""" if account is None: return self.authorization_backend.authorize_for_resource_type_action( identity_id, resource_type=permission.resource_type, action=permission.action, ) return self.authorization_backend.authorize_account( identity_id, resource_type=permission.resource_type, action=permission.action, account=account, ) def check_account_resource( self, identity_id: str, /, *, account: Account, permission: Permission, resource_id: str | None = None, joint_ventures: list[JointVentureParticipant] | None = None, ) -> None: """ Check if a user has a specific permission on a specific resource under account. """ allowed = self.authorization_backend.is_allowed_account_resource( identity_id, account=account, resource_id=resource_id or "0", resource_type=permission.resource_type, action=permission.action, joint_ventures=joint_ventures, ) self._raise_permission_denied(permission, allowed) def check_owned_resource( self, identity_id: str, /, *, permission: Permission, resource_id: str | None = None, ) -> None: """ Check if a user has a specific permission on a specific owned resource. """ allowed = self.authorization_backend.is_allowed_owned_resource( identity_id, resource_id=resource_id or "0", resource_identity_id=identity_id, resource_type=permission.resource_type, action=permission.action, account=None, ) self._raise_permission_denied(permission, allowed) def check_account_owned_resource( self, identity_id: str, /, *, account: Account, permission: Permission, resource_identity_id: str, resource_id: str | None = None, ) -> None: """ Check if a user has a specific permission on a specific owned resource. """ allowed = self.authorization_backend.is_allowed_owned_resource( identity_id, resource_id=resource_id or "0", resource_type=permission.resource_type, resource_identity_id=resource_identity_id, action=permission.action, account=account, ) self._raise_permission_denied(permission, allowed) @staticmethod def _raise_permission_denied(permission: Permission, allowed: bool) -> None: if not allowed: raise PermissionDenied( f"You are not allowed to perform `{permission.action}` action " f"on `{permission.resource_type}` resource." ) def get_account_access_for_permission( self, identity_id: str, /, *, permission: Permission, ) -> AccountAccess: """ Get account access for a specific resource type without an error if the account access is empty """ return self.authorization_backend.get_account_access( identity_id, resource_type=permission.resource_type, action=permission.action, )