pieter/eveAI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- Partner model additions

Browse Source 
- menu changes to allow for partners
- partner views and forms now in partner_forms.py and partner_views.py
- Introduction of services layer
- Allow all configuration to handle partner configurations, and adaptation of caching to allow for this
This commit is contained in:
Josako
2025-05-02 13:10:59 +02:00
parent 9652d0bff9
commit 6ef025363d
72 changed files with 1342 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
common/services/entitlement_services.py Normal file
View File

@@ -0,0 +1,68 @@
from flask import session, current_app, flash
from sqlalchemy.exc import SQLAlchemyError
from common.extensions import db
from common.models.entitlements import PartnerServiceLicenseTier
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 EntitlementServices:
@staticmethod
def associate_license_tier_with_partner(license_tier_id):
"""Associate a license tier 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:
flash("Cannot associate license tier with partner. No management service defined for partner", "danger")
current_app.logger.error(f"No Management Service defined for partner {partner_id}"
f"trying to associate license tier {license_tier_id}.")
raise EveAINoManagementPartnerService()
# Check if the association already exists
existing_association = PartnerServiceLicenseTier.query.filter_by(
partner_service_id=management_service['id'],
license_tier_id=license_tier_id
).first()
if existing_association:
# Association already exists, nothing to do
flash("License tier was already associated with partner", "info")
current_app.logger.info(f"Association between partner service {management_service['id']} and "
f"license tier {license_tier_id} already exists.")
return
# Create the association
association = PartnerServiceLicenseTier(
partner_service_id=management_service['id'],
license_tier_id=license_tier_id
)
set_logging_information(association, dt.now(tz.utc))
db.session.add(association)
db.session.commit()
flash("Successfully associated license tier to partner", "success")
current_app.logger.info(f"Successfully associated license tier {license_tier_id} with "
f"partner service {management_service['id']}")
return True
except SQLAlchemyError as e:
db.session.rollback()
flash("Failed to associated license tier with partner service due to an internal error. "
"Please contact the System Administrator", "danger")
current_app.logger.error(f"Error associating license tier {license_tier_id} with partner: {str(e)}")
raise e

47
common/services/partner_services.py Normal file
View File

@@ -0,0 +1,47 @@
from typing import List
from flask import session
from sqlalchemy.exc import SQLAlchemyError
from common.models.entitlements import PartnerServiceLicenseTier
from common.utils.eveai_exceptions import EveAINoManagementPartnerService, EveAINoSessionPartner
from common.utils.security_utils import current_user_has_role
class PartnerServices:
@staticmethod
def get_allowed_license_tier_ids() -> List[int]:
"""
Retrieve IDs of all License Tiers associated with the partner's management service
Returns:
List of license tier IDs
Raises:
EveAINoSessionPartner: If no partner is in the session
EveAINoManagementPartnerService: If partner has no management service
"""
partner = session.get("partner", None)
if not partner:
raise EveAINoSessionPartner()
# 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:
raise EveAINoManagementPartnerService()
management_service_id = management_service['id']
# Query for all license tiers associated with this management service
associations = PartnerServiceLicenseTier.query.filter_by(
partner_service_id=management_service_id
).all()
# Extract the license tier IDs
license_tier_ids = [assoc.license_tier_id for assoc in associations]
return license_tier_ids

71
common/services/tenant_service.py
View File

@@ -1,71 +0,0 @@
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

176
common/services/tenant_services.py Normal file
View File

@@ -0,0 +1,176 @@
from typing import Dict, List
from flask import session, current_app
from sqlalchemy.exc import SQLAlchemyError
from common.extensions import db, cache_manager
from common.models.user import Partner, PartnerTenant, PartnerService, Tenant
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 TenantServices:
@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 get_available_types_for_tenant(tenant_id: int, config_type: str) -> Dict[str, Dict[str, str]]:
"""
Get available configuration types for a tenant based on partner relationships
Args:
tenant_id: The tenant ID
config_type: The configuration type ('specialists', 'agents', 'tasks', etc.)
Returns:
Dictionary of available types for the tenant
"""
# Get the appropriate cache handler based on config_type
cache_handler = None
if config_type == 'specialists':
cache_handler = cache_manager.specialists_types_cache
elif config_type == 'agents':
cache_handler = cache_manager.agents_types_cache
elif config_type == 'tasks':
cache_handler = cache_manager.tasks_types_cache
elif config_type == 'tools':
cache_handler = cache_manager.tools_types_cache
else:
raise ValueError(f"Unsupported config type: {config_type}")
# Get all types with their metadata (including partner info)
all_types = cache_handler.get_types()
# Filter to include:
# 1. Types with no partner (global)
# 2. Types with partners that have a SPECIALIST_SERVICE relationship with this tenant
available_partners = TenantServices.get_tenant_partner_names(tenant_id)
available_types = {
type_id: info for type_id, info in all_types.items()
if info.get('partner') is None or info.get('partner') in available_partners
}
return available_types
@staticmethod
def get_tenant_partner_names(tenant_id: int) -> List[str]:
"""
Get names of partners that have a SPECIALIST_SERVICE relationship with this tenant
Args:
tenant_id: The tenant ID
Returns:
List of partner names (tenant names)
"""
# Find all PartnerTenant relationships for this tenant
partner_names = []
try:
# Get all partner services of type SPECIALIST_SERVICE
specialist_services = (
PartnerService.query
.filter_by(type='SPECIALIST_SERVICE')
.all()
)
if not specialist_services:
return []
# Find tenant relationships with these services
partner_tenants = (
PartnerTenant.query
.filter_by(tenant_id=tenant_id)
.filter(PartnerTenant.partner_service_id.in_([svc.id for svc in specialist_services]))
.all()
)
# Get the partner names (their tenant names)
for pt in partner_tenants:
partner_service = (
PartnerService.query
.filter_by(id=pt.partner_service_id)
.first()
)
if partner_service:
partner = Partner.query.get(partner_service.partner_id)
if partner:
# Get the tenant associated with this partner
partner_tenant = Tenant.query.get(partner.tenant_id)
if partner_tenant:
partner_names.append(partner_tenant.name)
except SQLAlchemyError as e:
current_app.logger.error(f"Database error retrieving partner names: {str(e)}")
return partner_names
@staticmethod
def can_use_specialist_type(tenant_id: int, specialist_type: str) -> bool:
"""
Check if a tenant can use a specific specialist type
Args:
tenant_id: The tenant ID
specialist_type: The specialist type ID
Returns:
True if the tenant can use the specialist type, False otherwise
"""
# Get the specialist type definition
try:
specialist_types = cache_manager.specialists_types_cache.get_types()
specialist_def = specialist_types.get(specialist_type)
if not specialist_def:
return False
# If it's a global specialist, anyone can use it
if specialist_def.get('partner') is None:
return True
# If it's a partner-specific specialist, check if tenant has access
partner_name = specialist_def.get('partner')
available_partners = TenantServices.get_tenant_partner_names(tenant_id)
return partner_name in available_partners
except Exception as e:
current_app.logger.error(f"Error checking specialist type access: {str(e)}")
return False

39
common/services/user_service.py
View File

@@ -1,39 +0,0 @@
from flask import session
from common.models.user import Partner, Role
# common/services/user_service.py
from common.utils.eveai_exceptions import EveAIRoleAssignmentException
from common.utils.security_utils import current_user_has_role, all_user_roles
class UserService:
@staticmethod
def get_assignable_roles():
"""Retrieves roles that can be assigned to a user depending on the current user logged in,
and the active tenant for the session"""
current_tenant_id = session.get('tenant').get('id', None)
effective_role_names = []
if current_tenant_id == 1:
if current_user_has_role("Super User"):
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") 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
@staticmethod
def validate_role_assignments(role_ids):
"""Validate a set of role assignments, raising exception for first invalid role"""
assignable_roles = UserService.get_assignable_roles()
assignable_role_ids = {role[0] for role in assignable_roles}
role_id_set = set(role_ids)
return role_id_set.issubset(assignable_role_ids)

95
common/services/user_services.py Normal file
View File

@@ -0,0 +1,95 @@
from flask import session
from common.models.user import Partner, Role, PartnerTenant
from common.utils.eveai_exceptions import EveAIRoleAssignmentException
from common.utils.security_utils import current_user_has_role
class UserServices:
@staticmethod
def get_assignable_roles():
"""Retrieves roles that can be assigned to a user depending on the current user logged in,
and the active tenant for the session"""
current_tenant_id = session.get('tenant').get('id', None)
effective_role_names = []
if current_tenant_id == 1:
if current_user_has_role("Super User"):
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") 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
@staticmethod
def validate_role_assignments(role_ids):
"""Validate a set of role assignments, raising exception for first invalid role"""
assignable_roles = UserServices.get_assignable_roles()
assignable_role_ids = {role[0] for role in assignable_roles}
role_id_set = set(role_ids)
return role_id_set.issubset(assignable_role_ids)
@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 = session.get('partner', None)
if partner and partner["tenant_id"] == tenant_id:
return True
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
@staticmethod
def can_user_create_tenant() -> 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_permissions = partner_service.get('permissions', None)
return partner_permissions.get('can_create_tenant', False)
else:
return False
@staticmethod
def can_user_assign_license() -> 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_permissions = partner_service.get('permissions', None)
return partner_permissions.get('can_assign_license', False)
else:
return False