Files
Odoo-Modules/fusion_api/models/api_provider.py
gsinghpal e56974d46f update
2026-03-16 08:14:56 -04:00

174 lines
6.4 KiB
Python

# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
from odoo import models, fields, api
class FusionApiProvider(models.Model):
_name = 'fusion.api.provider'
_description = 'API Service Provider'
_order = 'sequence, name'
name = fields.Char(required=True)
provider_type = fields.Selection([
('openai', 'OpenAI'),
('anthropic', 'Anthropic'),
('google_maps', 'Google Maps'),
('google_oauth', 'Google OAuth'),
('microsoft_oauth', 'Microsoft OAuth'),
('twilio', 'Twilio'),
('custom', 'Custom'),
], required=True, default='custom')
status = fields.Selection([
('active', 'Active'),
('inactive', 'Inactive'),
('error', 'Error'),
], default='inactive', required=True, tracking=True)
description = fields.Text()
website_url = fields.Char(string='API Dashboard URL')
sequence = fields.Integer(default=10)
color = fields.Integer()
icon_class = fields.Char(
string='Icon CSS Class',
help='Font Awesome class for display, e.g. fa-brain',
)
key_ids = fields.One2many('fusion.api.key', 'provider_id', string='API Keys')
access_ids = fields.One2many('fusion.api.access', 'provider_id', string='Access Rules')
active_key_count = fields.Integer(compute='_compute_key_stats', string='Active Keys')
total_month_cost = fields.Float(compute='_compute_usage_stats', string='Month Cost (USD)')
total_month_requests = fields.Integer(compute='_compute_usage_stats', string='Month Requests')
company_id = fields.Many2one('res.company', default=lambda self: self.env.company)
@api.depends('key_ids', 'key_ids.is_active')
def _compute_key_stats(self):
for rec in self:
rec.active_key_count = len(rec.key_ids.filtered('is_active'))
def _compute_usage_stats(self):
today = fields.Date.today()
month_start = today.replace(day=1)
for rec in self:
usages = self.env['fusion.api.usage'].sudo().search([
('provider_id', '=', rec.id),
('create_date', '>=', fields.Datetime.to_string(month_start)),
('status', '=', 'success'),
])
rec.total_month_cost = sum(u.estimated_cost_usd for u in usages)
rec.total_month_requests = len(usages)
def action_activate(self):
self.ensure_one()
if not self.key_ids.filtered('is_active'):
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'title': 'No Active Keys',
'message': 'Add at least one active API key before activating this provider.',
'type': 'warning',
'sticky': False,
},
}
self.status = 'active'
def action_deactivate(self):
self.ensure_one()
self.status = 'inactive'
@api.model
def get_dashboard_data(self):
today = fields.Date.today()
month_start = today.replace(day=1)
providers = self.search([])
consumers = self.env['fusion.api.consumer'].search([])
month_usages = self.env['fusion.api.usage'].sudo().search([
('create_date', '>=', fields.Datetime.to_string(month_start)),
('status', '=', 'success'),
])
today_usages = month_usages.filtered(
lambda u: u.create_date.date() == today
)
month_cost = sum(u.estimated_cost_usd for u in month_usages)
month_requests = len(month_usages)
today_requests = len(today_usages)
consumer_stats = {}
for usage in month_usages:
cid = usage.consumer_id.id
if cid not in consumer_stats:
consumer_stats[cid] = {
'name': usage.consumer_id.name,
'cost': 0.0,
'requests': 0,
}
consumer_stats[cid]['cost'] += usage.estimated_cost_usd
consumer_stats[cid]['requests'] += 1
top_consumers = sorted(
consumer_stats.values(),
key=lambda x: x['cost'],
reverse=True,
)[:5]
provider_stats = []
for prov in providers.filtered(lambda p: p.status == 'active'):
prov_usages = month_usages.filtered(lambda u: u.provider_id.id == prov.id)
provider_stats.append({
'id': prov.id,
'name': prov.name,
'type': prov.provider_type,
'cost': sum(u.estimated_cost_usd for u in prov_usages),
'requests': len(prov_usages),
'keys': prov.active_key_count,
})
recent = self.env['fusion.api.usage'].sudo().search(
[], limit=10, order='create_date desc',
)
recent_list = [{
'consumer': r.consumer_id.name or '',
'provider': r.provider_id.name or '',
'feature': r.feature or '',
'cost': round(r.estimated_cost_usd, 6),
'status': r.status,
'tokens': r.total_tokens,
'time': fields.Datetime.to_string(r.create_date),
} for r in recent]
approaching_limits = []
access_rules = self.env['fusion.api.access'].sudo().search([
('monthly_budget_usd', '>', 0),
('is_enabled', '=', True),
])
for rule in access_rules:
pct = rule.budget_usage_pct
if pct >= 80:
approaching_limits.append({
'consumer': rule.consumer_id.name,
'provider': rule.provider_id.name,
'pct': round(pct, 1),
'budget': rule.monthly_budget_usd,
'spent': round(rule.current_month_cost, 2),
})
return {
'total_providers': len(providers),
'active_providers': len(providers.filtered(lambda p: p.status == 'active')),
'total_consumers': len(consumers),
'active_consumers': len(consumers.filtered('is_active')),
'month_cost': round(month_cost, 2),
'month_requests': month_requests,
'today_requests': today_requests,
'top_consumers': top_consumers,
'provider_stats': provider_stats,
'recent_usage': recent_list,
'approaching_limits': approaching_limits,
}