svc-user-auth-zxw 3.0.0

用户权限验证工具包

svc_user_auth_zxw - 用户权限验证工具包

一个基于 FastAPI 的用户认证和权限管理系统，支持多种登录方式、JWT 认证、会员系统和权限控制。

版本信息

📈 版本历史

最新版本

非兼容性更新

• 3.0.0 🎉 重大更新:
  • 安全漏洞修复
  • 敏感权限操作由api改为内部函数

非兼容性更新

• 2.1.0 🎉 重大更新: 新增完整的会员体系功能
  • 添加会员类型管理（创建、查询、更新、删除）
  • 添加用户会员管理（购买、续费、状态管理）
  • 添加会员权限验证装饰器和API
  • 添加定时任务自动清理过期会员
  • 会员权限与角色权限系统无缝整合
  • ⚠️ 非兼容性更新，与之前版本不兼容

近期版本

• 2.0.4: 添加uniApp手机号登录页(具备完备的api请求逻辑, 复制到项目内直接使用)
• 2.0.3: 阿里云sms发送增加logger记录
• 2.0.2: 阿里云sms发送配置使用config中配置, 为None时使用默认配置
• 2.0.1: 修正使用文档错误
• 2.0.0: 项目由 user_auth_zxw 更名为 svc_user_auth_zxw

1.0.x 版本（适配VUE-ELEMENT-PLUS-ADMIN框架）

• 1.0.5: 用户文档错误修正
• 1.0.4: 添加用户文档
• 1.0.3: bug fix，新用户注册后to_payload()报错修复，优化用户角色关联数据加载
• 1.0.2: 优化账号密码注册登录返回值，变更数据表结构：Role采用联合主键：(app_id, name)
• 1.0.1: 新增vue-element-plus-admin api: 退出登录
• 1.0.0: 统一API返回值，符合vue-element-plus-admin框架原生标准

早期版本

• 0.1.1: 支持多线程任务，集成修改: 短信验证码验证，redis存储与验证
• 0.1.0: 修改: jwt验证失败, 弹出401 HTTPException_AppToolsSZXW异常
• 0.0.9: 表结构User新增字段:referer_id,referer,invitees，手机号注册登录新增相应字段，增加邀请人信息
• 0.0.8: 新增:批量删除用户角色(delete_roles)

功能特性

🔐 用户认证

• JWT Token 认证
• 多种登录方式：账号密码、手机短信、微信公众号
• Token 刷新机制
• 安全的密码加密

👥 权限管理

• 基于角色的权限控制（RBAC）
• 多应用权限隔离
• 权限验证装饰器
• 动态角色分配和删除
• 自动创建应用和角色
• 批量权限管理

💎 会员系统

• 定期会员功能
• 会员类型管理
• 基于会员的权限控制
• 自动过期处理

⏰ 定时任务

• 自动清理过期会员
• 会员统计报表
• 定时任务调度

快速开始

1. 安装

通过 pip 安装

pip install svc_user_auth_zxw

从源码安装

git clone [项目地址]
cd svc_user_auth_zxw
pip install -e .

2. 环境配置

数据库配置

在 configs/config_user_auth.py 中配置数据库连接：

# PostgreSQL 数据库
DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/database"

# Redis 数据库  
REDIS_URL_AUTH = "redis://:password@localhost:6379/0"

或通过环境变量：

export USER_CENTER_DATABASE_URL="postgresql+asyncpg://user:password@localhost:5432/database"
export REDIS_URL_AUTH="redis://:password@localhost:6379/0"

JWT 配置

class JWT:
    SECRET_KEY = "your-secret-key"  # 生产环境请使用强密钥
    ALGORITHM = "HS256"
    expire_time = timedelta(seconds=30)  # token过期时间

短信服务配置（阿里云）

class AliyunSMS:
    ali_access_key_id = "your-access-key-id"
    ali_access_key_secret = "your-access-key-secret"
    ali_secretNo_pool_key = "your-pool-key"

微信登录配置

class WeChatPub:
    app_id = "your-wechat-app-id"
    app_secret = "your-wechat-app-secret"

3. 快速启动

方式一：直接运行

python main.py

方式二：集成到现有 FastAPI 应用

from fastapi import FastAPI
from svc_user_auth_zxw.main import router
from contextlib import asynccontextmanager
from svc_user_auth_zxw.tools.scheduler import start_membership_scheduler, stop_membership_scheduler

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时
    await start_membership_scheduler()  # 启动会员定时任务
    yield
    # 关闭时
    await stop_membership_scheduler()

app = FastAPI(lifespan=lifespan)
app.include_router(router, prefix="/user_center")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8101)

4. 验证安装

服务启动后访问：

• 服务地址：http://localhost:8101
• API 文档：http://localhost:8101/docs
• 健康检查：http://localhost:8101/user_center/health

API 接口详细说明

1. JWT Token 管理 (/user_center/token)

获取当前用户信息

• 接口: POST /user_center/token/get-current-user/
• 认证: 需要 JWT Token
• 输入: 无（从 Header 中获取 Token）
• 输出:

  {
    "code": 200,
    "data": {
      "sub": "username",
      "username": "username", 
      "nickname": "昵称",
      "roles": [
        {
          "role_name": "admin",
          "app_name": "myapp",
          "app_id": 1
        }
      ]
    }
  }
  

刷新 Token

• 接口: POST /user_center/token/refresh-token/
• 输入:

  {
    "refresh_token": "refresh_token_string"
  }
  

• 输出:

  {
    "code": 200,
    "data": {
      "access_token": "new_access_token",
      "refresh_token": "new_refresh_token"
    }
  }
  

验证 Token（Body）

• 接口: POST /user_center/token/check-token-from-body/
• 输入:

  {
    "access_token": "access_token_string"
  }
  

• 输出: 返回解码后的 Payload

验证 Token（Header）

• 接口: POST /user_center/token/check-token-from-header/
• 认证: 需要在 Header 中传递 Authorization: Bearer token
• 输出: 返回解码后的 Payload

2. 账号密码注册登录 (/user_center/account)

注册

• 接口: POST /user_center/account/register/
• 输入:

  {
    "username": "testuser",
    "password": "password123",
    "role_name": "l0",
    "app_name": "app0"
  }
  

• 输出:

  {
    "code": 200,
    "data": {
      "access_token": "jwt_token",
      "refresh_token": "refresh_token",
      "user_info": {
        "sub": "testuser",
        "username": "testuser",
        "nickname": null,
        "roles": [...]
      }
    }
  }
  

登录

• 接口: POST /user_center/account/login/
• 输入:

  {
    "username": "testuser",
    "password": "password123"
  }
  

• 输出: 同注册接口

表单登录

• 接口: POST /user_center/account/login-form/
• 输入: 表单数据（application/x-www-form-urlencoded）

  username=testuser&password=password123
  

• 输出: 同注册接口

3. 手机注册登录 (/user_center/account/phone)

发送短信验证码

• 接口: POST /user_center/account/phone/send-verification-code/
• 输入: phone=13800138000 (Query 参数)
• 输出:

  {
    "code": 200,
    "data": {
      "message": "验证码已发送"
    }
  }
  

手机注册

• 接口: POST /user_center/account/phone/register-phone/
• 输入:

  {
    "phone": "13800138000",
    "sms_code": "1234",
    "email": "[e-mail]",
    "role_name": "l0",
    "app_name": "app0",
    "referer_id": null
  }
  

• 输出: 登录成功信息

手机登录

• 接口: POST /user_center/account/phone/login-phone/
• 输入:

  {
    "phone": "13800138000",
    "sms_code": "1234"
  }
  

• 输出: 登录成功信息

注册或登录（一键）

• 接口: POST /user_center/account/phone/register-or-login-phone/
• 功能: 自动判断用户是否存在，存在则登录，不存在则注册
• 输入: 同手机登录
• 输出: 登录成功信息

更换绑定手机号

• 接口: POST /user_center/account/phone/change-phone/
• 认证: 需要 JWT Token
• 输入: new_phone 和 sms_code (Query 参数)
• 输出: 更新后的用户信息

4. 微信公众号登录 (/user_center/wechat)

获取登录二维码

• 接口: POST /user_center/wechat/qr-login/get-qrcode
• 输入:

  {
    "WECHAT_REDIRECT_URI": "https://your-domain.com/callback"
  }
  

• 输出:

  {
    "qr_code_url": "https://open.weixin.qq.com/connect/qrconnect?..."
  }
  

微信登录

• 接口: POST /user_center/wechat/qr-login/login/
• 输入:

  {
    "code": "wechat_auth_code",
    "app_name": "myapp"
  }
  

• 输出: 登录成功信息

5. 权限管理 (/user_center/api/roles)

分配角色

• 接口: POST /user_center/api/roles/assign-role/
• 输入:

  {
    "user_id": 1,
    "role_name": "admin",
    "app_name": "myapp"
  }
  

• 输出:

  {
    "status": true,
    "message": "角色分配成功"
  }
  

验证角色权限

• 接口: POST /user_center/api/roles/role-auth/
• 认证: 需要 JWT Token
• 输入:

  {
    "role_name": "admin",
    "app_name": "myapp"
  }
  

• 输出:

  {
    "status": true
  }
  

Admin 权限示例

• 接口: GET /user_center/api/roles/admin-data/
• 认证: 需要 Admin 角色
• 输出: 管理员专属数据

6. 会员类型管理 (/user_center/membership-types)

获取会员类型列表

• 接口: GET /user_center/membership-types/
• 参数: skip=0&limit=100
• 输出:

  {
    "code": 200,
    "data": [
      {
        "id": 1,
        "name": "VIP会员",
        "description": "VIP会员享受所有高级功能",
        "duration_days": 30,
        "price": 9900,
        "is_active": true,
        "created_at": "2024-01-01T00:00:00Z"
      }
    ],
    "message": "获取成功"
  }
  

获取会员类型详情

• 接口: GET /user_center/membership-types/{membership_type_id}
• 输出: 单个会员类型详情

创建会员类型（管理员）

• 接口: POST /user_center/membership-types/admin/
• 认证: 需要管理员权限
• 输入:

  {
    "name": "超级VIP",
    "description": "超级VIP会员",
    "duration_days": 90,
    "price": 29900,
    "role_names": ["super_vip", "premium_access"],
    "app_name": "myapp"
  }
  

• 输出: 创建的会员类型信息

更新会员类型（管理员）

• 接口: PUT /user_center/membership-types/admin/{membership_type_id}
• 认证: 需要管理员权限
• 输入: 会员类型更新数据（字段可选）
• 输出: 更新后的会员类型信息

删除会员类型（管理员）

• 接口: DELETE /user_center/membership-types/admin/{membership_type_id}
• 认证: 需要管理员权限
• 输出: 删除确认信息

7. 用户会员管理 (/user_center/memberships)

获取我的会员

• 接口: GET /user_center/memberships/my
• 认证: 需要 JWT Token
• 输出:

  {
    "code": 200,
    "data": [
      {
        "id": 1,
        "user_id": 1,
        "membership_type_id": 1,
        "membership_type_name": "VIP会员",
        "start_time": "2024-01-01T00:00:00Z",
        "end_time": "2024-01-31T00:00:00Z",
        "status": "active",
        "created_at": "2024-01-01T00:00:00Z",
        "updated_at": "2024-01-01T00:00:00Z",
        "notes": "购买记录",
        "is_valid": true
      }
    ],
    "message": "获取成功"
  }
  

获取会员详情

• 接口: GET /user_center/memberships/{membership_id}
• 认证: 需要 JWT Token（只能查看自己的或管理员查看所有）
• 输出: 单个会员记录详情

管理员获取用户会员

• 接口: GET /user_center/memberships/admin/user/{user_id}
• 认证: 需要管理员权限
• 输出: 指定用户的所有会员记录

管理员更新会员状态

• 接口: PUT /user_center/memberships/admin/{membership_id}/status
• 认证: 需要管理员权限
• 输入:

  {
    "status": "suspended",
    "notes": "用户违规，暂停会员"
  }
  

• 输出: 更新后的会员信息

管理员获取所有活跃会员

• 接口: GET /user_center/memberships/admin/active/all
• 认证: 需要管理员权限
• 参数: skip=0&limit=100
• 输出: 所有活跃会员列表

管理员获取所有过期会员

• 接口: GET /user_center/memberships/admin/expired/all
• 认证: 需要管理员权限
• 参数: skip=0&limit=100
• 输出: 所有过期会员列表

8. 会员权限验证 (/user_center/membership-auth)

验证会员权限

• 接口: POST /user_center/membership-auth/verify
• 认证: 需要 JWT Token
• 输入:

  {
    "membership_type_name": "VIP会员",
    "role_name": "vip_user",
    "app_name": "myapp"
  }
  

• 输出:

  {
    "code": 200,
    "data": {
      "has_membership": true,
      "has_permission": true,
      "membership_info": {
        // 会员详细信息
      }
    },
    "message": "验证成功"
  }
  

获取我的有效会员

• 接口: GET /user_center/membership-auth/my-memberships
• 认证: 需要 JWT Token
• 输出: 当前用户的所有有效会员列表

9. 登出 (/user_center/logout)

登出

• 接口: POST /user_center/logout/
• 认证: 需要 JWT Token
• 功能: 使当前 refresh_token 失效
• 输出: 登出成功确认

可外部调用的核心函数

说明: 以下函数是可以在您的代码中直接导入和调用的内部函数，区别于上述的 REST API 接口。这些函数通常用于服务端的业务逻辑处理，无需通过 HTTP 请求调用。

1. JWT 相关函数

get_current_user(token: str) -> User

• 位置: svc_user_auth_zxw.SDK_jwt.jwt
• 作用: 从 JWT Token 获取当前用户
• 输入: JWT Token 字符串
• 输出: User 对象
• 使用示例:

  from svc_user_auth_zxw.SDK_jwt.jwt import get_current_user
  from fastapi import Depends
  
  @app.get("/protected")
  async def protected_route(user: User = Depends(get_current_user)):
      return {"user": user.username}
  

create_jwt_token(payload: dict) -> str

• 位置: svc_user_auth_zxw.SDK_jwt.jwt
• 作用: 创建 JWT Token
• 输入: 用户信息字典
• 输出: JWT Token 字符串

check_jwt_token(token: str) -> Payload

• 位置: svc_user_auth_zxw.SDK_jwt.jwt
• 作用: 验证并解码 JWT Token
• 输入: JWT Token 字符串
• 输出: Payload 对象或 None

2. 权限管理函数

add_new_role(request: 请求_分配或创建角色, db: AsyncSession)

• 位置: svc_user_auth_zxw.apis.api_用户权限_增加
• 作用: 给用户分配角色，如果角色或应用不存在则自动创建
• 输入: 分配角色请求对象、数据库会话
• 输出: 操作状态和消息
• 功能:
  • 如果用户不存在则返回404
  • 如果角色不存在则创建角色
  • 如果应用不存在则创建应用
  • 自动关联用户和角色
• 使用示例:

  from svc_user_auth_zxw.apis.api_用户权限_增加 import add_new_role
  from svc_user_auth_zxw.apis.schemas import 请求_分配或创建角色
  
  request = 请求_分配或创建角色(
      user_id=1,
      role_name="admin",
      app_name="myapp"
  )
  result = await add_new_role(request, db)
  

delete_role(user_id: int, role_name: str, db: AsyncSession)

• 位置: svc_user_auth_zxw.apis.api_用户权限_增加
• 作用: 解除用户与指定角色的关联
• 输入: 用户ID、角色名、数据库会话
• 输出: 操作状态和消息
• 使用示例:

  from svc_user_auth_zxw.apis.api_用户权限_增加 import delete_role
  
  result = await delete_role(user_id=1, role_name="admin", db=db)
  

delete_roles(user_id: int, role_names: list[str], db: AsyncSession)

• 位置: svc_user_auth_zxw.apis.api_用户权限_增加
• 作用: 批量解除用户与多个角色的关联
• 输入: 用户ID、角色名列表、数据库会话
• 输出: 操作状态和删除的角色数量
• 使用示例:

  from svc_user_auth_zxw.apis.api_用户权限_增加 import delete_roles
  
  result = await delete_roles(user_id=1, role_names=["admin", "user"], db=db)
  

3. 权限验证函数

require_role(role_name: str, app_name: str = None)

• 位置: svc_user_auth_zxw.apis.api_用户权限_验证
• 作用: 装饰器，验证用户是否具有指定角色
• 输入: 角色名、应用名（可选）
• 输出: 用户对象或抛出权限异常
• 使用示例:

  from svc_user_auth_zxw.apis.api_用户权限_验证 import require_role
  
  @app.get("/admin-only")
  async def admin_only(user: User = Depends(require_role("admin", "myapp"))):
      return {"message": "管理员专用接口"}
  

4. 会员权限验证函数

require_membership(membership_type_name: str, app_name: str = None)

• 位置: svc_user_auth_zxw.apis.api_会员权限验证
• 作用: 装饰器，验证用户是否具有有效会员
• 输入: 会员类型名、应用名（可选）
• 输出: 用户对象或抛出权限异常
• 使用示例:

  from svc_user_auth_zxw.apis.api_会员权限验证 import require_membership
  
  @app.get("/vip-content")
  async def vip_content(user: User = Depends(require_membership("VIP会员"))):
      return {"content": "VIP专享内容"}
  

require_membership_or_role(membership_type_name: str, role_name: str, app_name: str = None)

• 位置: svc_user_auth_zxw.apis.api_会员权限验证
• 作用: 装饰器，验证用户是否具有会员权限或角色权限（任一即可）
• 输入: 会员类型名、角色名、应用名（可选）
• 输出: 用户对象或抛出权限异常

5. 内部会员管理函数

internal_purchase_membership()

• 位置: svc_user_auth_zxw.apis.api_用户会员管理
• 作用: 内部会员购买函数（仅供内部调用）
• 输入: 用户ID、会员类型ID、验证状态等
• 输出: 会员响应对象
• 安全性: 受保护的内部函数，需要特定权限

internal_admin_grant_membership()

• 位置: svc_user_auth_zxw.apis.api_用户会员管理
• 作用: 管理员内部赠送会员函数
• 输入: 管理员用户、目标用户ID、会员类型ID等
• 输出: 会员响应对象
• 权限: 需要管理员权限

6. 数据库相关函数

get_db() -> AsyncSession

• 位置: svc_user_auth_zxw.db.get_db
• 作用: 获取数据库会话
• 输入: 无
• 输出: AsyncSession 对象
• 使用示例:

  from svc_user_auth_zxw.db.get_db import get_db
  
  @app.get("/users")
  async def get_users(db: AsyncSession = Depends(get_db)):
      # 数据库操作
      pass
  

7. 工具函数

SMS 验证码相关

• 类: SMSCodeChecker
• 位置: svc_user_auth_zxw.tools.check_sms_code
• 方法:
  • store_verification_code(phone: str, code: str): 存储验证码
  • verify_code(phone: str, code: str) -> bool: 验证验证码

定时任务管理

• 函数: start_membership_scheduler() / stop_membership_scheduler()
• 位置: svc_user_auth_zxw.tools.scheduler
• 作用: 启动/停止会员定时任务

使用示例

基础认证示例

from fastapi import FastAPI, Depends
from svc_user_auth_zxw.SDK_jwt.jwt import get_current_user
from svc_user_auth_zxw.db.models import User

app = FastAPI()

@app.get("/profile")
async def get_profile(user: User = Depends(get_current_user)):
    return {
        "username": user.username,
        "nickname": user.nickname,
        "roles": [{"name": role.name, "app": role.app.name} for role in user.roles]
    }

权限控制示例

from svc_user_auth_zxw.apis.api_用户权限_验证 import require_role

@app.get("/admin-dashboard")
async def admin_dashboard(user: User = Depends(require_role("admin"))):
    return {"message": "欢迎管理员", "user": user.username}

权限管理示例

from svc_user_auth_zxw.apis.api_用户权限_增加 import add_new_role, delete_role, delete_roles
from svc_user_auth_zxw.apis.schemas import 请求_分配或创建角色
from svc_user_auth_zxw.db.get_db import get_db

async def assign_user_role():
    """给用户分配角色示例"""
    db = get_db()
    
    # 分配管理员角色
    request = 请求_分配或创建角色(
        user_id=1,
        role_name="admin",
        app_name="myapp"
    )
    result = await add_new_role(request, db)
    print(f"分配结果: {result}")

async def remove_user_role():
    """删除用户角色示例"""
    db = get_db()
    
    # 删除单个角色
    result1 = await delete_role(user_id=1, role_name="admin", db=db)
    print(f"删除单个角色: {result1}")
    
    # 批量删除角色
    result2 = await delete_roles(user_id=1, role_names=["user", "editor"], db=db)
    print(f"批量删除角色: {result2}")

会员权限示例

from svc_user_auth_zxw.apis.api_会员权限验证 import require_membership

@app.get("/premium-feature")
async def premium_feature(user: User = Depends(require_membership("VIP会员"))):
    return {"feature": "高级功能", "user": user.username}

部署建议

1. 生产环境配置

• 使用强密钥和安全的数据库密码
• 配置 HTTPS
• 设置合适的 CORS 策略
• 配置日志收集

2. 数据库优化

• 为高频查询字段添加索引
• 定期备份用户数据和会员数据
• 监控数据库性能

3. 安全考虑

• 定期更新依赖包
• 监控异常登录行为
• 实施 API 访问频率限制
• 定期审查权限配置

故障排除

常见问题

1. Token 验证失败

   • 检查 JWT SECRET_KEY 配置
   • 确认 Token 是否过期
   • 验证 Header 格式是否正确

2. 数据库连接失败

   • 检查数据库 URL 配置
   • 确认数据库服务是否运行
   • 验证用户权限

3. 短信发送失败

   • 检查阿里云 SMS 配置
   • 确认账户余额充足
   • 验证短信模板是否正确

4. 会员权限验证失败

   • 检查会员是否过期
   • 确认会员类型配置正确
   • 验证角色关联设置

日志查看

• 应用日志：logs/app.log
• API 日志：logs/svc_user_auth_zxw.apis.*.log

接口与函数使用说明

API 接口 vs 内部函数

API 接口（REST API）

• 使用场景: 前端应用、移动端、第三方系统调用
• 调用方式: 通过 HTTP 请求（GET、POST、PUT、DELETE）
• 认证方式: 需要 JWT Token 或其他认证方式
• 示例: POST /user_center/api/roles/assign-role/

内部函数（Python 函数）

• 使用场景: 服务端内部业务逻辑、其他 Python 模块调用
• 调用方式: 直接导入并调用 Python 函数
• 数据库操作: 需要传递数据库会话对象
• 示例: await add_new_role(request, db)

选择建议

• 前端开发: 使用 REST API 接口
• 后端集成: 优先使用内部函数，性能更好
• 微服务调用: 使用 REST API 接口
• 数据库事务: 使用内部函数，便于事务管理

更新日志

v2.1.0

• 新增定期会员功能
• 优化安全性
• 完善文档
• 补充权限管理内部函数说明

联系方式

• 作者: 景募
• 邮箱: [e-mail]
• 许可证: MIT License

---

注意: 本文档基于 v2.1.0 版本编写，使用前请确保版本匹配。生产环境部署前请仔细配置安全参数。