Underground-Digital/Workflow-Engine

1import os2 3import requests4 5from extensions.ext_database import db6from models.account import TenantAccountJoin, TenantAccountRole7 8 9class BillingService:10    base_url = os.environ.get("BILLING_API_URL", "BILLING_API_URL")11    secret_key = os.environ.get("BILLING_API_SECRET_KEY", "BILLING_API_SECRET_KEY")12 13    @classmethod14    def get_info(cls, tenant_id: str):15        params = {"tenant_id": tenant_id}16 17        billing_info = cls._send_request("GET", "/subscription/info", params=params)18 19        return billing_info20 21    @classmethod22    def get_subscription(cls, plan: str, interval: str, prefilled_email: str = "", tenant_id: str = ""):23        params = {"plan": plan, "interval": interval, "prefilled_email": prefilled_email, "tenant_id": tenant_id}24        return cls._send_request("GET", "/subscription/payment-link", params=params)25 26    @classmethod27    def get_model_provider_payment_link(cls, provider_name: str, tenant_id: str, account_id: str, prefilled_email: str):28        params = {29            "provider_name": provider_name,30            "tenant_id": tenant_id,31            "account_id": account_id,32            "prefilled_email": prefilled_email,33        }34        return cls._send_request("GET", "/model-provider/payment-link", params=params)35 36    @classmethod37    def get_invoices(cls, prefilled_email: str = "", tenant_id: str = ""):38        params = {"prefilled_email": prefilled_email, "tenant_id": tenant_id}39        return cls._send_request("GET", "/invoices", params=params)40 41    @classmethod42    def _send_request(cls, method, endpoint, json=None, params=None):43        headers = {"Content-Type": "application/json", "Billing-Api-Secret-Key": cls.secret_key}44 45        url = f"{cls.base_url}{endpoint}"46        response = requests.request(method, url, json=json, params=params, headers=headers)47 48        return response.json()49 50    @staticmethod51    def is_tenant_owner_or_admin(current_user):52        tenant_id = current_user.current_tenant_id53 54        join = (55            db.session.query(TenantAccountJoin)56            .filter(TenantAccountJoin.tenant_id == tenant_id, TenantAccountJoin.account_id == current_user.id)57            .first()58        )59 60        if not TenantAccountRole.is_privileged_role(join.role):61            raise ValueError("Only team owner or team admin can perform this action")62