296 lines
13 KiB
Python
296 lines
13 KiB
Python
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 sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy import or_, desc
|
|
import ast
|
|
|
|
from common.models.entitlements import License, LicenseTier, LicenseUsage, BusinessEventLog
|
|
from common.extensions import db, security, minio_client, simple_encryption
|
|
from .entitlements_forms import LicenseTierForm, LicenseForm
|
|
from common.utils.view_assistants import prepare_table_for_macro, form_validation_failed
|
|
from common.utils.nginx_utils import prefixed_url_for
|
|
|
|
entitlements_bp = Blueprint('entitlements_bp', __name__, url_prefix='/entitlements')
|
|
|
|
|
|
@entitlements_bp.route('/license_tier', methods=['GET', 'POST'])
|
|
@roles_accepted('Super User')
|
|
def license_tier():
|
|
form = LicenseTierForm()
|
|
if form.validate_on_submit():
|
|
current_app.logger.info("Adding License Tier")
|
|
|
|
new_license_tier = LicenseTier()
|
|
form.populate_obj(new_license_tier)
|
|
|
|
try:
|
|
db.session.add(new_license_tier)
|
|
db.session.commit()
|
|
except SQLAlchemyError as e:
|
|
db.session.rollback()
|
|
current_app.logger.error(f'Failed to add license tier to database. Error: {str(e)}')
|
|
flash(f'Failed to add license tier to database. Error: {str(e)}', 'success')
|
|
return render_template('entitlements/license_tier.html', form=form)
|
|
|
|
current_app.logger.info(f"Successfully created license tier {new_license_tier.id}")
|
|
flash(f"Successfully created tenant license tier {new_license_tier.id}", 'success')
|
|
|
|
return redirect(prefixed_url_for('entitlements_bp.view_license_tiers'))
|
|
else:
|
|
form_validation_failed(request, form)
|
|
|
|
return render_template('entitlements/license_tier.html', form=form)
|
|
|
|
|
|
@entitlements_bp.route('/view_license_tiers', methods=['GET', 'POST'])
|
|
@roles_required('Super User')
|
|
def view_license_tiers():
|
|
page = request.args.get('page', 1, type=int)
|
|
per_page = request.args.get('per_page', 10, type=int)
|
|
today = dt.now(tz.utc)
|
|
|
|
query = LicenseTier.query.filter(
|
|
or_(
|
|
LicenseTier.end_date == None,
|
|
LicenseTier.end_date >= today
|
|
)
|
|
).order_by(LicenseTier.start_date.desc(), LicenseTier.id)
|
|
|
|
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
|
|
license_tiers = pagination.items
|
|
|
|
rows = prepare_table_for_macro(license_tiers, [('id', ''), ('name', ''), ('version', ''), ('start_date', ''),
|
|
('end_date', '')])
|
|
|
|
return render_template('entitlements/view_license_tiers.html', rows=rows, pagination=pagination)
|
|
|
|
|
|
@entitlements_bp.route('/handle_license_tier_selection', methods=['POST'])
|
|
@roles_required('Super User')
|
|
def handle_license_tier_selection():
|
|
license_tier_identification = request.form['selected_row']
|
|
license_tier_id = ast.literal_eval(license_tier_identification).get('value')
|
|
the_license_tier = LicenseTier.query.get(license_tier_id)
|
|
|
|
action = request.form['action']
|
|
|
|
match action:
|
|
case 'edit_license_tier':
|
|
return redirect(prefixed_url_for('entitlements_bp.edit_license_tier',
|
|
license_tier_id=license_tier_id))
|
|
case 'create_license_for_tenant':
|
|
return redirect(prefixed_url_for('entitlements_bp.create_license',
|
|
license_tier_id=license_tier_id))
|
|
# Add more conditions for other actions
|
|
return redirect(prefixed_url_for('entitlements_bp.view_license_tiers'))
|
|
|
|
|
|
@entitlements_bp.route('/license_tier/<int:license_tier_id>', methods=['GET', 'POST'])
|
|
@roles_accepted('Super User')
|
|
def edit_license_tier(license_tier_id):
|
|
license_tier = LicenseTier.query.get_or_404(license_tier_id) # This will return a 404 if no license tier is found
|
|
form = LicenseTierForm(obj=license_tier)
|
|
|
|
if form.validate_on_submit():
|
|
# Populate the license_tier with form data
|
|
form.populate_obj(license_tier)
|
|
|
|
try:
|
|
db.session.add(license_tier)
|
|
db.session.commit()
|
|
except SQLAlchemyError as e:
|
|
db.session.rollback()
|
|
current_app.logger.error(f'Failed to edit License Tier. Error: {str(e)}')
|
|
flash(f'Failed to edit License Tier. Error: {str(e)}', 'danger')
|
|
return render_template('entitlements/license_tier.html', form=form, license_tier_id=license_tier.id)
|
|
|
|
flash('License Tier updated successfully.', 'success')
|
|
return redirect(
|
|
prefixed_url_for('entitlements_bp.edit_license_tier', license_tier_id=license_tier_id))
|
|
else:
|
|
form_validation_failed(request, form)
|
|
|
|
return render_template('entitlements/license_tier.html', form=form, license_tier_id=license_tier.id)
|
|
|
|
|
|
@entitlements_bp.route('/create_license/<int:license_tier_id>', methods=['GET', 'POST'])
|
|
@roles_accepted('Super User')
|
|
def create_license(license_tier_id):
|
|
form = LicenseForm()
|
|
tenant_id = session.get('tenant').get('id')
|
|
currency = session.get('tenant').get('currency')
|
|
|
|
if request.method == 'GET':
|
|
# Fetch the LicenseTier
|
|
license_tier = LicenseTier.query.get_or_404(license_tier_id)
|
|
|
|
# Prefill the form with LicenseTier data
|
|
# Currency depending data
|
|
if currency == '$':
|
|
form.basic_fee.data = license_tier.basic_fee_d
|
|
form.additional_storage_price.data = license_tier.additional_storage_price_d
|
|
form.additional_embedding_price.data = license_tier.additional_embedding_price_d
|
|
form.additional_interaction_token_price.data = license_tier.additional_interaction_token_price_d
|
|
elif currency == '€':
|
|
form.basic_fee.data = license_tier.basic_fee_e
|
|
form.additional_storage_price.data = license_tier.additional_storage_price_e
|
|
form.additional_embedding_price.data = license_tier.additional_embedding_price_e
|
|
form.additional_interaction_token_price.data = license_tier.additional_interaction_token_price_e
|
|
else:
|
|
current_app.logger.error(f'Invalid currency {currency} for tenant {tenant_id} while creating license.')
|
|
flash(f"Invalid currency {currency} for tenant {tenant_id} while creating license. "
|
|
f"Check tenant's currency and try again.", 'danger')
|
|
return redirect(prefixed_url_for('user_bp.edit_tenant', tenant_id=tenant_id))
|
|
# General data
|
|
form.currency.data = currency
|
|
form.max_storage_mb.data = license_tier.max_storage_mb
|
|
form.additional_storage_bucket.data = license_tier.additional_storage_bucket
|
|
form.included_embedding_mb.data = license_tier.included_embedding_mb
|
|
form.additional_embedding_bucket.data = license_tier.additional_embedding_bucket
|
|
form.included_interaction_tokens.data = license_tier.included_interaction_tokens
|
|
form.additional_interaction_bucket.data = license_tier.additional_interaction_bucket
|
|
form.overage_embedding.data = license_tier.standard_overage_embedding
|
|
form.overage_interaction.data = license_tier.standard_overage_interaction
|
|
else: # POST
|
|
# Create a new License instance
|
|
new_license = License(
|
|
tenant_id=tenant_id,
|
|
tier_id=license_tier_id,
|
|
)
|
|
if form.validate_on_submit():
|
|
# Update the license with form data
|
|
form.populate_obj(new_license)
|
|
# Currency is added here again, as a form doesn't include disabled fields when passing it in the request
|
|
new_license.currency = currency
|
|
|
|
try:
|
|
db.session.add(new_license)
|
|
db.session.commit()
|
|
flash('License created successfully', 'success')
|
|
return redirect(prefixed_url_for('entitlements_bp.edit_license', license_id=new_license.id))
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
flash(f'Error creating license: {str(e)}', 'error')
|
|
else:
|
|
form_validation_failed(request, form)
|
|
|
|
return render_template('entitlements/license.html', form=form, ext_disabled_fields=[])
|
|
|
|
|
|
@entitlements_bp.route('/license/<int:license_id>', methods=['GET', 'POST'])
|
|
@roles_accepted('Super User')
|
|
def edit_license(license_id):
|
|
license = License.query.get_or_404(license_id) # This will return a 404 if no license tier is found
|
|
form = LicenseForm(obj=license)
|
|
disabled_fields = []
|
|
if len(license.usages) > 0: # There already are usage records linked to this license
|
|
# Define which fields should be disabled
|
|
disabled_fields = [field.name for field in form if field.name != 'end_date']
|
|
|
|
if form.validate_on_submit():
|
|
# Populate the license with form data
|
|
form.populate_obj(license)
|
|
|
|
try:
|
|
db.session.add(license)
|
|
db.session.commit()
|
|
except SQLAlchemyError as e:
|
|
db.session.rollback()
|
|
current_app.logger.error(f'Failed to edit License. Error: {str(e)}')
|
|
flash(f'Failed to edit License. Error: {str(e)}', 'danger')
|
|
return render_template('entitlements/license.html', form=form)
|
|
|
|
flash('License updated successfully.', 'success')
|
|
return redirect(
|
|
prefixed_url_for('entitlements_bp.edit_license', license_id=license_id))
|
|
else:
|
|
form_validation_failed(request, form)
|
|
|
|
return render_template('entitlements/license.html', form=form, ext_disabled_fields=disabled_fields)
|
|
|
|
|
|
@entitlements_bp.route('/view_usages')
|
|
@roles_accepted('Super User', 'Tenant Admin')
|
|
def view_usages():
|
|
page = request.args.get('page', 1, type=int)
|
|
per_page = request.args.get('per_page', 10, type=int)
|
|
|
|
tenant_id = session.get('tenant').get('id')
|
|
query = LicenseUsage.query.filter_by(tenant_id=tenant_id).order_by(desc(LicenseUsage.id))
|
|
|
|
pagination = query.paginate(page=page, per_page=per_page)
|
|
lus = pagination.items
|
|
|
|
# prepare table data
|
|
|
|
rows = prepare_table_for_macro(lus, [('id', ''), ('period_start_date', ''), ('period_end_date', ''),
|
|
('storage_mb_used', ''), ('embedding_mb_used', ''),
|
|
('interaction_total_tokens_used', '')])
|
|
|
|
# Render the users in a template
|
|
return render_template('entitlements/view_usages.html', rows=rows, pagination=pagination)
|
|
|
|
|
|
@entitlements_bp.route('/handle_usage_selection', methods=['POST'])
|
|
@roles_accepted('Super User', 'Tenant Admin')
|
|
def handle_usage_selection():
|
|
usage_identification = request.form['selected_row']
|
|
usage_id = ast.literal_eval(usage_identification).get('value')
|
|
the_usage = LicenseUsage.query.get_or_404(usage_id)
|
|
|
|
action = request.form['action']
|
|
|
|
pass # Currently, no actions are defined
|
|
|
|
|
|
@entitlements_bp.route('/view_licenses')
|
|
@roles_accepted('Super User', 'Tenant Admin')
|
|
def view_licenses():
|
|
page = request.args.get('page', 1, type=int)
|
|
per_page = request.args.get('per_page', 10, type=int)
|
|
|
|
tenant_id = session.get('tenant').get('id')
|
|
# Get current date in UTC
|
|
current_date = dt.now(tz=tz.utc).date()
|
|
|
|
# Query licenses for the tenant, with ordering and active status
|
|
query = (
|
|
License.query
|
|
.join(LicenseTier) # Join with LicenseTier
|
|
.filter(License.tenant_id == tenant_id)
|
|
.add_columns(
|
|
License.id,
|
|
License.start_date,
|
|
License.end_date,
|
|
LicenseTier.name.label('license_tier_name'), # Access name through LicenseTier
|
|
((License.start_date <= current_date) &
|
|
(or_(License.end_date.is_(None), License.end_date >= current_date))).label('active')
|
|
)
|
|
.order_by(License.start_date.desc())
|
|
)
|
|
|
|
pagination = query.paginate(page=page, per_page=per_page)
|
|
lics = pagination.items
|
|
|
|
# prepare table data
|
|
rows = prepare_table_for_macro(lics, [('id', ''), ('license_tier_name', ''), ('start_date', ''), ('end_date', ''),
|
|
('active', '')])
|
|
|
|
# Render the licenses in a template
|
|
return render_template('entitlements/view_licenses.html', rows=rows, pagination=pagination)
|
|
|
|
|
|
@entitlements_bp.route('/handle_license_selection', methods=['POST'])
|
|
@roles_accepted('Super User', 'Tenant Admin')
|
|
def handle_license_selection():
|
|
license_identification = request.form['selected_row']
|
|
license_id = ast.literal_eval(license_identification).get('value')
|
|
the_license = License.query.get_or_404(license_id)
|
|
|
|
action = request.form['action']
|
|
|
|
match action:
|
|
case 'edit_license':
|
|
return redirect(prefixed_url_for('entitlements_bp.edit_license', license_id=license_id)) |