# -*- coding:utf-8 -*- import re from sanic.request import RequestParameters from functools import wraps from sanic.response import json from pot_libs.aiohttp_util.aiohttp_utils import AioHttpUtils from pot_libs.settings import SETTING from unify_api.modules.users.procedures import jwt_utils from pot_libs.logger import log from jwt.compat import binary_type, text_type from pot_libs.aredis_util import aredis_utils from sanic import response from unify_api.modules.users.components.logout_cps import ( LogoutRequest, LogoutResponse ) from pot_libs.sanic_api import summary, description from pot_libs.common.components.responses import Success from unify_api.modules.users.procedures import user_product_auth def authorized(): def decorator(f): @wraps(f) async def decorated_function(request, *args, **kwargs): # run some method that checks the request # for the client's authorization status is_authorized = await user_product_auth.check_user_auth(request) if is_authorized: # the user is authorized. # run the handler method and return the response response = await f(request, *args, **kwargs) return response else: # the user is not authorized. return json({'status': 'not_authorized'}, 403) return decorated_function return decorator @authorized() async def get_test(request): return json({'status': 'authorized'}) @summary('登出') async def get_user_logout(request) -> Success: token = request.token if token: # await jwt_utils.store_token_blacklist(token) try: log.info(f"request logout_url={SETTING.logout_url}") request_body = { "db": SETTING.mysql_db } resp_str, status = await AioHttpUtils().post( SETTING.logout_url, request_body, timeout=50, headers={"Authorization": f"Bearer {token}"}, ) log.info(f"request auth_url resp_str={resp_str} status={status}") if status != 200: return Success(0) else: return Success(1) except Exception as e: log.exception(e) return Success(0) else: log.error("header token is missing!") return Success(0) @summary('logout') @description('user logout store token to token_backlist') async def post_assistant(request, body: LogoutRequest) -> LogoutResponse: sid = body.sid values = []