logout.py 2.6 KB
Newer Older
lcn's avatar
lcn committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56
# -*- 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}")
lcn's avatar
lcn committed
57 58 59 60
            request_body = {
                "db": SETTING.mysql_db
            }
            resp_str, status = await AioHttpUtils().post(
lcn's avatar
lcn committed
61
                SETTING.logout_url,
lcn's avatar
lcn committed
62 63
                request_body,
                timeout=50,
lcn's avatar
lcn committed
64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83
                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 = []