from collections.abc import Callable from functools import wraps from typing import Optional from flask import request from flask_restful import reqparse from pydantic import BaseModel from extensions.ext_database import db from models.account import Tenant def get_tenant(view: Optional[Callable] = None): def decorator(view_func): @wraps(view_func) def decorated_view(*args, **kwargs): # fetch json body parser = reqparse.RequestParser() parser.add_argument('tenant_id', type=str, required=True, location='json') parser.add_argument('user_id', type=str, required=True, location='json') kwargs = parser.parse_args() user_id = kwargs.get('user_id') tenant_id = kwargs.get('tenant_id') del kwargs['tenant_id'] del kwargs['user_id'] try: tenant_model = db.session.query(Tenant).filter( Tenant.id == tenant_id, ).first() except Exception: raise ValueError('tenant not found') if not tenant_model: raise ValueError('tenant not found') kwargs['tenant_model'] = tenant_model kwargs['user_id'] = user_id return view_func(*args, **kwargs) return decorated_view if view is None: return decorator else: return decorator(view) def plugin_data(view: Optional[Callable] = None, *, payload_type: type[BaseModel]): def decorator(view_func): def decorated_view(*args, **kwargs): try: data = request.get_json() except Exception: raise ValueError('invalid json') try: payload = payload_type(**data) except Exception as e: raise ValueError(f'invalid payload: {str(e)}') kwargs['payload'] = payload return view_func(*args, **kwargs) return decorated_view if view is None: return decorator else: return decorator(view)