import ast import uuid from datetime import datetime as dt, timezone as tz from flask import request, redirect, flash, render_template, Blueprint, session, current_app, jsonify from flask_security import hash_password, roles_required, roles_accepted, current_user from itsdangerous import URLSafeTimedSerializer from sqlalchemy.exc import SQLAlchemyError from common.extensions import db, cache_manager from common.models.user import Partner, Tenant, PartnerService, PartnerTenant from common.utils.celery_utils import current_celery from common.utils.eveai_exceptions import EveAIException from common.utils.log_utils import format_query_results from common.utils.model_logging_utils import update_logging_information, set_logging_information from common.utils.view_assistants import prepare_table_for_macro, form_validation_failed from common.utils.nginx_utils import prefixed_url_for from .partner_forms import TriggerActionForm, EditPartnerForm, PartnerServiceForm, EditPartnerServiceForm partner_bp = Blueprint('partner_bp', __name__, url_prefix='/partner') @partner_bp.route('/trigger_actions', methods=['GET']) @roles_accepted('Super User') def trigger_actions(): form = TriggerActionForm() return render_template('partner/trigger_actions.html', form=form) @partner_bp.route('/handle_trigger_action', methods=['POST']) @roles_accepted('Super User') def handle_trigger_action(): action = request.form['action'] match action: case 'update_usages': try: # Use send_task to trigger the task since it's part of another component (eveai_entitlements) task = current_celery.send_task('update_usages', queue='entitlements') current_app.logger.info(f"Usage update task triggered: {task.id}") flash('Usage update task has been triggered successfully!', 'success') except Exception as e: current_app.logger.error(f"Failed to trigger usage update task: {str(e)}") flash(f'Failed to trigger usage update: {str(e)}', 'danger') return redirect(prefixed_url_for('partner_bp.trigger_actions')) # Partner Management ------------------------------------------------------------------------------ @partner_bp.route('/partner/', methods=['GET', 'POST']) @roles_accepted('Super User') def edit_partner(partner_id): partner = Partner.query.get_or_404(partner_id) # This will return a 404 if no partner is found tenant = Tenant.query.get_or_404(partner.tenant_id) form = EditPartnerForm(obj=partner) if request.method == 'GET': form.tenant.data = tenant.name if form.validate_on_submit(): current_app.logger.debug(f"Form data for Partner: {form.data}") # Populate the user with form data form.populate_obj(partner) update_logging_information(partner, dt.now(tz.utc)) db.session.commit() flash('Partner updated successfully.', 'success') refresh_session_partner(partner.id) return redirect( prefixed_url_for('partner_bp.edit_partner', partner_id=partner.id)) # Assuming there's a user profile view to redirect to else: form_validation_failed(request, form) return render_template('partner/edit_partner.html', form=form, partner_id=partner_id) @partner_bp.route('/partners', methods=['GET', 'POST']) @roles_accepted('Super User') def partners(): page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) query = (db.session.query( Partner.id, Partner.code, Partner.active, Partner.logo_url, # Include all needed Partner columns here Tenant.name.label('name') ).join(Tenant, Partner.tenant_id == Tenant.id).order_by(Partner.id)) current_app.logger.debug(f'{format_query_results(query)}') pagination = query.paginate(page=page, per_page=per_page) the_partners = pagination.items # prepare table data rows = prepare_table_for_macro(the_partners, [('id', ''), ('name', '')]) # Render the catalogs in a template return render_template('partner/partners.html', rows=rows, pagination=pagination) @partner_bp.route('/handle_partner_selection', methods=['POST']) @roles_accepted('Super User') def handle_partner_selection(): action = request.form['action'] if action == 'create_partner': try: partner_id = register_partner_from_tenant(session['tenant']['id']) return redirect(prefixed_url_for('partner_bp.edit_partner', partner_id=partner_id, )) except EveAIException as e: current_app.logger.error(f'Error registering partner for tenant {session['tenant']['id']}: {str(e)}') flash('Error Registering Partner for Selected Tenant', 'danger') return redirect(prefixed_url_for('partner_bp.partners')) partner_identification = request.form.get('selected_row') partner_id = ast.literal_eval(partner_identification).get('value') partner = Partner.query.get_or_404(partner_id) if action == 'set_session_partner': current_app.logger.info(f"Setting session partner: {partner.id}") session['partner'] = partner.to_dict() elif action == 'edit_partner': return redirect(prefixed_url_for('partner_bp.edit_partner', partner_id=partner_id)) return redirect(prefixed_url_for('partner_bp.partners')) # Partner Servide Management ---------------------------------------------------------------------- @partner_bp.route('/partner_service', methods=['GET', 'POST']) @roles_accepted('Super User') def partner_service(): form = PartnerServiceForm() if form.validate_on_submit(): partner = session.get('partner', None) if not partner: flash('No partner has been selected. Set partner before adding services.', 'warning') return redirect(prefixed_url_for('partner_bp.partners')) partner_id = partner['id'] new_partner_service = PartnerService() form.populate_obj(new_partner_service) set_logging_information(new_partner_service, dt.now(tz.utc)) new_partner_service.partner_id = partner_id try: db.session.add(new_partner_service) db.session.commit() flash('Partner Service successfully added!', 'success') current_app.logger.info(f"Partner Service {new_partner_service.name} added successfully for {partner_id}") # Step 2 of the creation process (depending on type) return redirect(prefixed_url_for('partner_bp.partner_service', partner_service_id=new_partner_service.id)) except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to add Partner Service: {str(e)}', 'danger') current_app.logger.error(f"Failed to add Partner Service {new_partner_service.name} " f"for partner {partner_id}. Error: {str(e)}") return render_template('partner/partner_service.html', form=form) @partner_bp.route('/partner_service/', methods=['GET', 'POST']) @roles_accepted('Super User') def edit_partner_service(partner_service_id): partner_service = PartnerService.query.get_or_404(partner_service_id) partner_id = session['partner']['id'] form = EditPartnerServiceForm(request.form, obj=partner_service) partner_service_config = cache_manager.partner_services_config_cache.get_config(partner_service.type, partner_service.type_version) form.add_dynamic_fields("configuration", partner_service_config, partner_service.configuration) form.add_dynamic_fields("permissions", partner_service_config, partner_service.permissions) if request.method == 'POST': current_app.logger.debug(f"Form returned: {form.data}") raw_form_data = request.form.to_dict() current_app.logger.debug(f"Raw form data: {raw_form_data}") if form.validate_on_submit(): form.populate_obj(partner_service) partner_service.configuration = form.get_dynamic_data('configuration') partner_service.permissions = form.get_dynamic_data('permissions') current_app.logger.debug(f"Partner Service configuration: {partner_service.configuration}") current_app.logger.debug(f"Partner Service permissions: {partner_service.permissions}") update_logging_information(partner_service, dt.now(tz.utc)) try: db.session.add(partner_service) db.session.commit() flash('Partner Service updated successfully.', 'success') current_app.logger.info(f"Partner Service {partner_service.name} updated successfully! ") refresh_session_partner(partner_id) except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to update Partner Service: {str(e)}', 'danger') current_app.logger.error(f"Failed to update Partner Service {partner_service.id} for partner {partner_id}. " f"Error: {str(e)} ") return render_template('partner/edit_partner_service.html', form=form, partner_service_id=partner_service_id) return redirect(prefixed_url_for('partner_bp.partner_services')) else: form_validation_failed(request, form) return render_template('partner/edit_partner_service.html', form=form, partner_service_id=partner_service_id) @partner_bp.route('/partner_services', methods=['GET', 'POST']) @roles_accepted('Super User') def partner_services(): page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) partner = session.get('partner', None) if not partner: flash('No partner has been selected. Set partner before adding services.', 'warning') return redirect(prefixed_url_for('partner_bp.partners')) partner_id = session['partner']['id'] query = PartnerService.query.filter(PartnerService.partner_id == partner_id) pagination = query.paginate(page=page, per_page=per_page) the_partner_services = pagination.items # prepare table data rows = prepare_table_for_macro(the_partner_services, [('id', ''), ('name', ''), ('type', '')]) return render_template('partner/partner_services.html', rows=rows, pagination=pagination) @partner_bp.route('/handle_partner_service_selection', methods=['POST']) @roles_accepted('Super User') def handle_partner_service_selection(): action = request.form['action'] if action == 'create_partner_service': return redirect(prefixed_url_for('partner_bp.partner_service')) partner_service_identification = request.form.get('selected_row') partner_service_id = ast.literal_eval(partner_service_identification).get('value') if action == 'edit_partner_service': return redirect(prefixed_url_for('partner_bp.edit_partner_service', partner_service_id=partner_service_id)) elif action == 'add_partner_service_for_tenant': add_partner_service_for_tenant(partner_service_id) return redirect(prefixed_url_for('partner_bp.edit_partner_service', partner_service_id=partner_service_id)) return redirect(prefixed_url_for('partner_bp.partner_services')) # Utility Functions def register_partner_from_tenant(tenant_id): # check if there is already a partner defined for the tenant partner = Partner.query.filter_by(tenant_id=tenant_id).first() if partner: return partner.id try: partner = Partner( tenant_id=tenant_id, code=f"PART-{str(uuid.uuid4())}", ) set_logging_information(partner, dt.now(tz.utc)) db.session.add(partner) db.session.commit() return partner.id except SQLAlchemyError as e: db.session.rollback() raise EveAIException(f"Failed to register partner for tenant {tenant_id}. Error: {str(e)}") def add_partner_service_for_tenant(partner_service_id): """ Associate a partner service with the current tenant Args: partner_service_id: ID of the partner service to associate Returns: Redirect to appropriate page based on result """ tenant = session.get('tenant', None) if not tenant: flash('No tenant has been selected. Set tenant before adding services.', 'warning') return redirect(prefixed_url_for('user_bp.tenants')) tenant_id = tenant['id'] try: # Check if the partner service exists partner_service = PartnerService.query.get(partner_service_id) if not partner_service: flash(f'Partner service with ID {partner_service_id} not found.', 'danger') return redirect(prefixed_url_for('partner_bp.partner_services')) # Check if the association already exists existing = PartnerTenant.query.filter_by( tenant_id=tenant_id, partner_service_id=partner_service_id ).first() if existing: flash(f'This tenant already has access to this partner service.', 'warning') return redirect(prefixed_url_for('partner_bp.partner_services')) # Create new association new_partner_tenant = PartnerTenant( tenant_id=tenant_id, partner_service_id=partner_service_id, # Add any additional fields needed for your model ) # Set logging information set_logging_information(new_partner_tenant, dt.now(tz.utc)) # Save to database db.session.add(new_partner_tenant) db.session.commit() # Get partner name for the flash message partner = Partner.query.get(partner_service.partner_id) partner_tenant = Tenant.query.get(partner.tenant_id) if partner else None partner_name = partner_tenant.name if partner_tenant else 'Unknown Partner' flash(f'Successfully added {partner_service.type} service from {partner_name} to this tenant.', 'success') return redirect(prefixed_url_for('partner_bp.partner_services')) except SQLAlchemyError as e: db.session.rollback() current_app.logger.error(f"Database error adding partner service: {str(e)}") flash(f'Error adding partner service: {str(e)}', 'danger') return redirect(prefixed_url_for('partner_bp.partner_services')) def refresh_session_partner(partner_id): if session.get('partner', None): if partner_id == session['partner']['id']: session['partner'] = Partner.query.get_or_404(partner_id).to_dict()