- Add 'Partner Admin' role to actual functionality in eveai_app
This commit is contained in:
71
common/services/tenant_service.py
Normal file
71
common/services/tenant_service.py
Normal file
@@ -0,0 +1,71 @@
|
||||
from flask import session, current_app
|
||||
from sqlalchemy.exc import SQLAlchemyError
|
||||
|
||||
from common.extensions import db
|
||||
from common.models.user import Partner, PartnerTenant
|
||||
from common.utils.eveai_exceptions import EveAINoManagementPartnerService
|
||||
from common.utils.model_logging_utils import set_logging_information
|
||||
from datetime import datetime as dt, timezone as tz
|
||||
|
||||
from common.utils.security_utils import current_user_has_role
|
||||
|
||||
|
||||
class TenantService:
|
||||
@staticmethod
|
||||
def associate_tenant_with_partner(tenant_id):
|
||||
"""Associate a tenant with a partner"""
|
||||
try:
|
||||
partner_id = session['partner']['id']
|
||||
# Get partner service (MANAGEMENT_SERVICE type)
|
||||
partner = Partner.query.get(partner_id)
|
||||
if not partner:
|
||||
return
|
||||
|
||||
# Find a management service for this partner
|
||||
management_service = next((service for service in session['partner']['services']
|
||||
if service.get('type') == 'MANAGEMENT_SERVICE'), None)
|
||||
|
||||
if not management_service:
|
||||
current_app.logger.error(f"No Management Service defined for partner {partner_id}"
|
||||
f"while associating tenant {tenant_id} with partner.")
|
||||
raise EveAINoManagementPartnerService()
|
||||
|
||||
# Create the association
|
||||
tenant_partner = PartnerTenant(
|
||||
partner_service_id=management_service['id'],
|
||||
tenant_id=tenant_id,
|
||||
relationship_type='MANAGED',
|
||||
)
|
||||
set_logging_information(tenant_partner, dt.now(tz.utc))
|
||||
|
||||
db.session.add(tenant_partner)
|
||||
db.session.commit()
|
||||
|
||||
except SQLAlchemyError as e:
|
||||
db.session.rollback()
|
||||
current_app.logger.error(f"Error associating tenant {tenant_id} with partner: {str(e)}")
|
||||
raise e
|
||||
|
||||
@staticmethod
|
||||
def can_user_edit_tenant(tenant_id) -> bool:
|
||||
if current_user_has_role('Super User'):
|
||||
return True
|
||||
elif current_user_has_role('Partner Admin'):
|
||||
partner_id = session['partner']['id']
|
||||
partner_service = next((service for service in session['partner']['services']
|
||||
if service.get('type') == 'MANAGEMENT_SERVICE'), None)
|
||||
if not partner_service:
|
||||
return False
|
||||
else:
|
||||
partner_tenant = PartnerTenant.query.filter(
|
||||
PartnerTenant.tenant_id == tenant_id,
|
||||
PartnerTenant.partner_service_id == partner_service['id'],
|
||||
).first()
|
||||
if partner_tenant:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
@@ -14,26 +14,21 @@ class UserService:
|
||||
and the active tenant for the session"""
|
||||
current_tenant_id = session.get('tenant').get('id', None)
|
||||
effective_role_names = []
|
||||
if current_tenant_id:
|
||||
if current_tenant_id == 1:
|
||||
if current_user_has_role("Super User"):
|
||||
if current_tenant_id == 1:
|
||||
effective_role_names.append("Super User")
|
||||
if session.get('partner'):
|
||||
effective_role_names.append("Partner Admin")
|
||||
effective_role_names.append("Tenant Admin")
|
||||
effective_role_names.append("Super User")
|
||||
elif current_tenant_id:
|
||||
if current_user_has_role("Tenant Admin"):
|
||||
effective_role_names.append("Tenant Admin")
|
||||
if current_user_has_role("Partner Admin"):
|
||||
if current_user_has_role("Partner Admin") or current_user_has_role("Super User"):
|
||||
effective_role_names.append("Tenant Admin")
|
||||
if session.get('partner'):
|
||||
if session.get('partner').get('tenant_id') == current_tenant_id:
|
||||
effective_role_names.append("Partner Admin")
|
||||
effective_role_names = list(set(effective_role_names))
|
||||
effective_roles = [(role.id, role.name) for role in
|
||||
Role.query.filter(Role.name.in_(effective_role_names)).all()]
|
||||
return effective_roles
|
||||
else:
|
||||
return []
|
||||
effective_role_names = list(set(effective_role_names))
|
||||
effective_roles = [(role.id, role.name) for role in
|
||||
Role.query.filter(Role.name.in_(effective_role_names)).all()]
|
||||
return effective_roles
|
||||
|
||||
@staticmethod
|
||||
def validate_role_assignments(role_ids):
|
||||
|
||||
Reference in New Issue
Block a user