| parser.add_argument('interval', type=str, required=True, location='args', choices=['month', 'year']) | parser.add_argument('interval', type=str, required=True, location='args', choices=['month', 'year']) | ||||
| args = parser.parse_args() | args = parser.parse_args() | ||||
| BillingService.is_tenant_owner(current_user) | |||||
| return BillingService.get_subscription(args['plan'], | return BillingService.get_subscription(args['plan'], | ||||
| args['interval'], | args['interval'], | ||||
| current_user.email, | current_user.email, | ||||
| @account_initialization_required | @account_initialization_required | ||||
| @only_edition_cloud | @only_edition_cloud | ||||
| def get(self): | def get(self): | ||||
| BillingService.is_tenant_owner(current_user) | |||||
| return BillingService.get_invoices(current_user.email) | return BillingService.get_invoices(current_user.email) | ||||
| import os | import os | ||||
| import requests | import requests | ||||
| from extensions.ext_database import db | |||||
| from models.account import TenantAccountJoin | |||||
| class BillingService: | class BillingService: | ||||
| base_url = os.environ.get('BILLING_API_URL', 'BILLING_API_URL') | base_url = os.environ.get('BILLING_API_URL', 'BILLING_API_URL') | ||||
| response = requests.request(method, url, json=json, params=params, headers=headers) | response = requests.request(method, url, json=json, params=params, headers=headers) | ||||
| return response.json() | return response.json() | ||||
| @staticmethod | |||||
| def is_tenant_owner(current_user): | |||||
| tenant_id = current_user.current_tenant_id | |||||
| join = db.session.query(TenantAccountJoin).filter( | |||||
| TenantAccountJoin.tenant_id == tenant_id, | |||||
| TenantAccountJoin.account_id == current_user.id | |||||
| ).first() | |||||
| if join.role != 'owner': | |||||
| raise ValueError('Only tenant owner can perform this action') |