- Introduction of Partner Admin role in combination with 'Management Partner' type.

This commit is contained in:
Josako
2025-04-09 09:40:59 +02:00
parent c2c3b01b28
commit f43e79376c
17 changed files with 368 additions and 111 deletions

View File

@@ -147,3 +147,10 @@ class EveAIDoublePartner(EveAIException):
message = f"Tenant with ID '{tenant_id}' is already defined as a Partner."
super().__init__(message, status_code, payload)
class EveAIRoleAssignmentException(EveAIException):
"""Exception raised when a role cannot be assigned due to business rules"""
def __init__(self, message, status_code=403, payload=None):
super().__init__(message, status_code, payload)

View File

@@ -1,7 +1,7 @@
from flask import session, current_app
from sqlalchemy import and_
from common.models.user import Tenant
from common.models.user import Tenant, Partner
from common.models.entitlements import License
from common.utils.database import Database
from common.utils.eveai_exceptions import EveAITenantNotFound, EveAITenantInvalid, EveAINoActiveLicense
@@ -13,13 +13,19 @@ def set_tenant_session_data(sender, user, **kwargs):
tenant = Tenant.query.filter_by(id=user.tenant_id).first()
session['tenant'] = tenant.to_dict()
session['default_language'] = tenant.default_language
session['default_llm_model'] = tenant.llm_model
partner = Partner.query.filter_by(tenant_id=user.tenant_id).first()
if partner:
session['partner'] = partner.to_dict()
else:
# Remove partner from session if it exists
session.pop('partner', None)
def clear_tenant_session_data(sender, user, **kwargs):
session.pop('tenant', None)
session.pop('default_language', None)
session.pop('default_llm_model', None)
session.pop('partner', None)
def is_valid_tenant(tenant_id):
@@ -40,4 +46,4 @@ def is_valid_tenant(tenant_id):
if not active_license:
raise EveAINoActiveLicense(tenant_id)
return True
return True

View File

@@ -1,8 +1,10 @@
from flask import current_app, render_template
from flask_security import current_user
from flask_mailman import EmailMessage
from itsdangerous import URLSafeTimedSerializer
import socket
from common.models.user import Role
from common.utils.nginx_utils import prefixed_url_for
@@ -93,3 +95,44 @@ def test_smtp_connection():
except Exception as e:
current_app.logger.error(f"Failed to connect to SMTP server: {str(e)}")
return False
def get_current_user_roles():
"""Get the roles of the currently authenticated user.
Returns:
List of Role objects or empty list if no user is authenticated
"""
if current_user.is_authenticated:
return current_user.roles
return []
def current_user_has_role(role_name):
"""Check if the current user has the specified role.
Args:
role_name (str): Name of the role to check
Returns:
bool: True if user has the role, False otherwise
"""
if not current_user.is_authenticated:
return False
return any(role.name == role_name for role in current_user.roles)
def current_user_roles():
"""Get the roles of the currently authenticated user.
Returns:
List of Role objects or empty list if no user is authenticated
"""
if current_user.is_authenticated:
return current_user.roles
return []
def all_user_roles():
roles = [(role.id, role.name) for role in Role.query.all()]