python-fastapi-mm-zhcs-yj-api/common/auth_user.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from exceptions import AuthException
import urllib
import hashlib
from fastapi import Depends, Request
from .security import valid_access_token
from .enc import mpfun
from extensions import * 
from database import get_db
from sqlalchemy.orm import Session
from utils.ry_system_util import user_id_get_user_info

class AuthUser:
    user_id: int
    user_name: str
    nick_name: str
    is_yzy_user: int

def data_auth_sign(data):
    sorted(data.items(),key=lambda item:item[0],reverse=True)
    res = urllib.parse.urlencode(data)
    sha = hashlib.sha1(res.encode('utf-8'))
    encrypts = sha.hexdigest()
    return encrypts

def get_auth_user(request: Request):
    try:
        auth = request.session.get('user_auth')
        if auth is not None:
            user_auth_sign = request.session['user_auth_sign']
            if user_auth_sign == data_auth_sign(auth):
                authUser = AuthUser()
                authUser.user_id = auth['user_id']
                authUser.user_name = auth['user_name']
                authUser.nick_name = auth['nick_name']
                authUser.is_yzy_user = int(auth['is_yzy_user'])
                return authUser
    except Exception as e:
        raise AuthException(str(e))
    return None

def find_auth_user(user_id: int = Depends(valid_access_token), db: Session = Depends(get_db)):
    try:
        user_info = user_id_get_user_info(db, user_id)
        if user_info is not None:
            authUser = AuthUser()
            authUser.user_id = str(user_info.user_id)
            authUser.user_name = mpfun.dec_data(user_info.user_name)
            authUser.nick_name = user_info.nick_name
            authUser.is_yzy_user = 0
            return authUser
    except Exception as e:
        raise AuthException(str(e))
    return None