import functools from pot_libs.utils.pendulum_wrapper import my_pendulum from unify_api.modules.ulock.dao.lock_dao import user_lock_auth_by_user_id async def check_user_lock_auth(request): args = request.json lock_id = args.get("id") user_id = request.ctx.user_id # user_id = 100197 lock_auth_list = await user_lock_auth_by_user_id(user_id) # 具备的lock_id权限, 如果包含all, 则有所有权限 lock_id_auth = [auth["lock_id"] for auth in lock_auth_list] if "all" in lock_id_auth or lock_id in lock_id_auth: return True return False def lock_required(): """用户开锁权限验证,装饰器""" # factory=2可用于装饰器括号后面传参 @lock_required(factory=5) def decorator(f, factory=2): @functools.wraps(f) async def wrapper(request, *args, **kwargs): is_lock_authed = await check_user_lock_auth(request) if is_lock_authed: return await f(request, *args, **kwargs) else: fail_body = { "code": 5002, "message": "用户没有该开锁权限", "data": {}, "srv_time": my_pendulum.now().to_datetime_string() } return [fail_body, 200] return wrapper return decorator