svc-user-auth-zxw 2.1.0

用户权限验证工具包

svc_user_auth_zxw 用户认证权限管理库

📚 简介

svc_user_auth_zxw 是一个基于 FastAPI 的用户认证和权限管理 Python 库，提供完整的用户注册、登录、权限验证等功能。支持多种登录方式，包括账号密码、微信登录、手机号验证码登录等。

✨ 核心特性

• 🔐 多种登录方式：账号密码、微信、手机验证码
• 🎯 灵活的角色权限管理系统
• 💎 完整的会员体系功能
  • 📋 会员类型管理（创建、查询、更新、删除）
  • 👥 用户会员管理（购买、续费、状态管理）
  • 🛡️ 会员权限验证装饰器
  • ⏰ 自动过期会员清理
  • 🔄 会员与角色权限整合
• 🔑 JWT Token 认证机制
• 📱 短信验证码支持
• 🗄️ Redis 缓存支持
• 🎨 适配 Vue-Element-Plus-Admin 框架
• 🚀 异步数据库操作（PostgreSQL）
• 🔄 Token 刷新机制

📋 目录

1. 版本历史
2. 安装
3. 快速开始
4. 核心API文档
5. 会员体系
6. 权限管理
7. 前端集成
8. 高级功能
9. 错误处理
10. 部署指南

📈 版本历史

最新版本

• 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)

🚀 安装

通过 pip 安装

pip install svc_user_auth_zxw

⚡ 快速开始

1. 项目配置

在项目根目录创建 configs/config_user_auth.py 文件：

import os
from datetime import timedelta

# PostgreSQL 数据库配置
DATABASE_URL = os.environ.get('USER_CENTER_DATABASE_URL')
if not DATABASE_URL:
    DATABASE_URL = "postgresql+asyncpg://username:password@localhost:5432/database_name"
    os.environ['USER_CENTER_DATABASE_URL'] = DATABASE_URL

# Redis 数据库配置
REDIS_URL_AUTH = os.environ.get('REDIS_URL_AUTH')
if not REDIS_URL_AUTH:
    REDIS_URL_AUTH = "redis://:your_redis_password@localhost:6379/0"
    os.environ['REDIS_URL_AUTH'] = REDIS_URL_AUTH

# JWT 配置
class JWT:
    SECRET_KEY = "your-secret-key-here"
    ALGORITHM = "HS256"
    expire_time = timedelta(seconds=3600)  # 1小时

# 应用配置
app_name = "your_app_name"

# 微信配置（可选）
class WeChatPub:
    app_id = "your_wechat_app_id"
    app_secret = "your_wechat_app_secret"
    scope = "snsapi_login"
    state = "your_custom_state"

# 阿里云短信配置（可选）
class AliyunSMS:
    ali_access_key_id = "your_access_key_id"
    ali_access_key_secret = "your_access_key_secret"

2. FastAPI 应用集成

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from svc_user_auth_zxw import router as user_auth_router
from svc_user_auth_zxw.tools.scheduler import start_membership_scheduler, stop_membership_scheduler

# 应用生命周期管理（包含会员体系定时任务）
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时
    print("🚀 启动用户认证服务...")
    await start_membership_scheduler()  # 启动会员清理定时任务
    print("⏰ 会员定时任务已启动")
    yield
    # 关闭时
    print("⏹️ 停止会员定时任务...")
    await stop_membership_scheduler()
    print("✅ 用户认证服务已关闭")

# 创建 FastAPI 应用
app = FastAPI(
    title="你的应用",
    description="基于 svc_user_auth_zxw 的用户认证系统",
    lifespan=lifespan
)

# 添加 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境请设置具体域名
    allow_credentials=True,
    allow_methods=["GET", "POST", "DELETE", "PUT", "OPTIONS"],
    allow_headers=["*"],
)

# 注册用户认证路由
app.include_router(user_auth_router, prefix="/user_center", tags=["用户认证"])

# 健康检查接口
@app.get("/")
async def root():
    return {"message": "用户认证服务运行正常", "version": "2.1.0"}

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

3. 数据库初始化

在项目主函数导入 from svc_user_auth_zxw import router 模块时，自动进行数据库初始化。

4. 基础使用示例

from fastapi import Depends
from svc_user_auth_zxw import get_current_user, require_role
from svc_user_auth_zxw.apis.api_会员权限验证 import require_membership

# 获取当前用户信息
@app.get("/profile")
async def get_profile(current_user=Depends(get_current_user)):
    return {"user": current_user.username, "roles": current_user.roles}

# 需要管理员权限的接口
@app.get("/admin-data")
async def admin_only(user=Depends(require_role("admin", "your_app"))):
    return {"message": "管理员专用数据", "user": user.username}

# 需要VIP会员权限的接口
@app.get("/vip-content")
async def vip_only(user=Depends(require_membership("VIP会员"))):
    return {"message": "VIP专享内容", "user": user.username}

📖 核心API文档

前缀说明: 所有API都需要加上前缀 /user_center（如果在集成时设置了prefix）

🔐 账号密码认证

1. 用户注册

• URL: /api/account/normal/register/
• 方法: POST

// 请求参数
{
    "username": "testuser",
    "password": "password123",
    "role_name": "user",      // 可选，默认 "l0"
    "app_name": "your_app"    // 可选，默认 "app0"
}

// 响应示例
{
    "code": 200,
    "data": {
        "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",
        "refresh_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",
        "user_info": {
            "username": "testuser",
            "nickname": "testuser",
            "roles": [...],
            "email": "testuser",
            "phone": "testuser"
        }
    }
}

2. 用户登录

• URL: /api/account/normal/login/
• 方法: POST

// 请求参数
{
    "username": "testuser",
    "password": "password123"
}

// 响应示例（同注册格式）
{
    "code": 200,
    "data": {
        "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",
        "refresh_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",
        "user_info": {
            "username": "testuser",
            "nickname": "testuser",
            "roles": [...],
            "email": "testuser",
            "phone": "testuser"
        }
    }
}

3. 表单登录

• URL: /api/account/normal/login-form/
• 方法: POST
• Content-Type: application/x-www-form-urlencoded

username=testuser&password=password123&grant_type=password

📱 手机号认证

1. 发送验证码

• URL: /api/account/phone/send-verification-code/
• 方法: POST

{
    "phone": "13800138000"
}

2. 手机号注册或登录（推荐）

• URL: /api/account/phone/register-or-login-phone/
• 方法: POST

// 请求参数
{
    "phone": "13800138000",
    "sms_code": "123456",
    "referer_id": 123,        // 可选，邀请人ID
    "app_name": "your_app"    // 可选
}

// 响应示例
{
    "code": 200,
    "data": {
        "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",
        "refresh_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",
        "user_info": {
            "username": "13800138000",
            "nickname": "13800138000",
            "roles": [...],
            "email": null,
            "phone": "13800138000"
        }
    }
}

3. 更换绑定手机号

• URL: /api/account/phone/change-phone/
• 方法: POST
• 需要认证: ✅

{
    "new_phone": "13900139000",
    "sms_code": "123456"
}

🔗 微信登录认证

1. 获取登录二维码URL

• URL: /api/account/wechat/qr-login/get-qrcode
• 方法: POST

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

2. 微信一键登录

• URL: /api/account/wechat/qr-login/login/
• 方法: POST

{
    "code": "微信返回的授权码",
    "app_name": "your_app"
}

🎫 JWT Token管理

1. 获取当前用户信息

• URL: /api/token/get-current-user/
• 方法: POST
• 需要认证: ✅

# Headers
Authorization: Bearer your_access_token

2. 刷新访问令牌

• URL: /api/token/refresh-token/
• 方法: POST

{
    "refresh_token": "your_refresh_token"
}

👥 用户权限管理API

1. 分配或创建角色

• URL: /api/roles/assign-role/
• 方法: POST

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

2. 验证用户角色

• URL: /api/roles/role-auth/
• 方法: POST
• 需要认证: ✅

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

🚪 用户登出

退出登录

• URL: /api/account/logout
• 方法: GET

💎 会员体系

🎯 会员体系概述

会员体系是 v2.1.0 的核心新功能，提供完整的时间控制、权限整合和自动化管理：

• ⏰ 时间控制: 会员在指定时间段内有效，自动检测和处理过期会员
• 🔄 权限整合: 与现有角色权限系统完美整合，支持组合使用
• 🤖 自动化管理: 定时任务自动清理过期会员，支持会员统计和监控
• 💰 灵活定价: 支持不同会员类型的差异化定价和时长

📊 核心数据结构

会员类型表 (membership_types)

CREATE TABLE membership_types (
    id SERIAL PRIMARY KEY,
    name VARCHAR(50) UNIQUE NOT NULL,
    description TEXT,
    duration_days INTEGER NOT NULL,
    price INTEGER NOT NULL,  -- 价格（分为单位）
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP DEFAULT NOW()
);

用户会员表 (memberships)

CREATE TABLE memberships (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    membership_type_id INTEGER REFERENCES membership_types(id),
    start_time TIMESTAMP NOT NULL,
    end_time TIMESTAMP NOT NULL,
    status VARCHAR(20) DEFAULT 'active',  -- active/expired/suspended/cancelled
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW(),
    notes TEXT
);

🔧 会员类型管理API

1. 创建会员类型

• URL: /membership-types/
• 方法: POST

// 请求参数
{
    "name": "VIP会员",
    "description": "VIP会员享受所有高级功能",
    "duration_days": 30,
    "price": 9900,  // 99元（以分为单位）
    "role_names": ["vip_user", "premium_access"],
    "app_name": "myapp"
}

// 响应示例
{
    "code": 200,
    "data": {
        "id": 1,
        "name": "VIP会员",
        "description": "VIP会员享受所有高级功能",
        "duration_days": 30,
        "price": 9900,
        "is_active": true,
        "created_at": "2024-12-20T10:00:00Z"
    },
    "message": "会员类型创建成功"
}

2. 获取会员类型列表

• URL: /membership-types/
• 方法: GET

3. 更新/删除会员类型

• URL: /membership-types/{membership_type_id}
• 方法: PUT / DELETE

👥 用户会员管理API

1. 购买会员

• URL: /memberships/purchase
• 方法: POST

// 请求参数
{
    "user_id": 1,
    "membership_type_id": 1,
    "notes": "购买VIP会员"
}

// 响应示例
{
    "code": 200,
    "data": {
        "id": 1,
        "user_id": 1,
        "membership_type_id": 1,
        "membership_type_name": "VIP会员",
        "start_time": "2024-12-20T10:00:00Z",
        "end_time": "2025-01-19T10:00:00Z",
        "status": "active",
        "is_valid": true
    },
    "message": "会员开通成功"
}

2. 获取我的会员

• URL: /memberships/my
• 方法: GET
• 需要认证: ✅

🛡️ 会员权限验证

权限装饰器使用

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

# 纯会员权限控制
@app.get("/vip-content")
async def get_vip_content(user=Depends(require_membership("VIP会员"))):
    return {"message": "这是VIP专享内容", "user": user.username}

# 会员+应用权限控制
@app.get("/app-vip-content")
async def get_app_vip_content(user=Depends(require_membership("VIP会员", "myapp"))):
    return {"message": "这是myapp应用中的VIP专享内容"}

# 会员或角色权限（任一即可）
@app.get("/premium-or-admin-content")
async def get_premium_or_admin_content(
    user=Depends(require_membership_or_role("Premium会员", "admin", "myapp"))
):
    return {"message": "Premium会员或admin角色可访问"}

API验证接口

• URL: /membership-auth/verify
• 方法: POST
• 需要认证: ✅

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

⏰ 定时任务和自动化管理

定时任务已集成到FastAPI应用的生命周期中，会自动执行以下操作：

• 🔄 每小时检查并清理过期会员
• 📊 统计会员使用情况
• 🚨 监控异常状态会员

手动管理工具

from svc_user_auth_zxw.tools.membership_cleanup import cleanup_expired_memberships

# 试运行（不实际修改数据）
result = await cleanup_expired_memberships(dry_run=True)

# 实际执行清理
result = await cleanup_expired_memberships(dry_run=False)

命令行工具

# 清理过期会员
python -m svc_user_auth_zxw.tools.membership_cleanup --cleanup

# 试运行模式
python -m svc_user_auth_zxw.tools.membership_cleanup --cleanup --dry-run

# 查看统计信息
python -m svc_user_auth_zxw.tools.membership_cleanup --stats

# 查看即将过期的会员（7天内）
python -m svc_user_auth_zxw.tools.membership_cleanup --expiring 7

🛡️ 权限管理

1. 权限体系概述

系统支持两种权限类型：

• 角色权限: 基于用户角色的传统权限管理
• 会员权限: 基于会员类型和时间的权限管理

2. 角色权限装饰器

from svc_user_auth_zxw import require_role, require_roles, get_current_user

# 单个角色验证
@app.get("/admin-only")
async def admin_endpoint(user=Depends(require_role("admin", "your_app"))):
    return {"message": "只有admin可以访问"}

# 多个角色验证（任一即可）
@app.get("/multi-roles")
async def multi_roles_endpoint(user=Depends(require_roles(["admin", "moderator"], "your_app"))):
    return {"message": "admin或moderator可以访问"}

# 获取当前用户
@app.get("/profile")
async def get_profile(current_user=Depends(get_current_user)):
    return {"user": current_user.username}

3. 权限组合使用

# 复杂的权限组合：必须是VIP会员且有admin角色
@app.get("/super-admin")
async def super_admin_endpoint(
    vip_user=Depends(require_membership("VIP会员")),
    admin_user=Depends(require_role("admin", "your_app"))
):
    # 只有同时满足VIP会员和admin角色的用户才能访问
    return {"message": "超级管理员专享功能"}

4. 动态权限管理

from svc_user_auth_zxw.apis.api_用户权限_增加 import add_new_role, delete_role, delete_roles

# 新增权限
await add_new_role(
    请求_分配或创建角色(
        user_id=user.id,
        role_name="gen_content_1",
        app_name="your_app"
    ),
    db=db
)

# 删除单个角色
await delete_role(user_id=1, role_name="admin", db=db_session)

# 批量删除角色
await delete_roles(user_id=1, role_names=["admin", "editor"], db=db_session)

🎨 前端集成

1. Vue-Element-Plus-Admin 框架集成

该库已适配 Vue-Element-Plus-Admin 框架的 API 标准格式：

// API 返回格式
{
    "code": 200,
    "data": {
        // 实际数据
    }
}

2. 前端登录示例

// 登录函数
async function login(username, password) {
    const response = await fetch('/user_center/api/account/normal/login/', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json',
        },
        body: JSON.stringify({ username, password })
    });
    
    const result = await response.json();
    if (result.code === 200) {
        localStorage.setItem('access_token', result.data.access_token);
        localStorage.setItem('refresh_token', result.data.refresh_token);
        return result.data.user_info;
    }
    throw new Error(result.data);
}

// 请求拦截器
axios.interceptors.request.use(config => {
    const token = localStorage.getItem('access_token');
    if (token) {
        config.headers.Authorization = `Bearer ${token}`;
    }
    return config;
});

// 响应拦截器（自动刷新Token）
axios.interceptors.response.use(
    response => response,
    async error => {
        if (error.response?.status === 401) {
            const refreshToken = localStorage.getItem('refresh_token');
            if (refreshToken) {
                try {
                    const response = await axios.post('/user_center/api/token/refresh-token/', {
                        refresh_token: refreshToken
                    });
                    const { access_token, refresh_token } = response.data.data;
                    localStorage.setItem('access_token', access_token);
                    localStorage.setItem('refresh_token', refresh_token);
                    // 重试原请求
                    return axios.request(error.config);
                } catch {
                    // 刷新失败，跳转到登录页
                    localStorage.clear();
                    window.location.href = '/login';
                }
            }
        }
        return Promise.reject(error);
    }
);

3. UniApp集成示例

项目提供了完整的UniApp手机号登录页面示例，位于 uniapp前端页面demo/ 目录。

🚀 高级功能

1. 短信验证码

库集成了阿里云短信服务，支持验证码发送和验证：

# 配置阿里云短信服务
class AliyunSMS:
    ali_access_key_id = "your_access_key_id"
    ali_access_key_secret = "your_access_key_secret"
    sign_name = "你的签名"
    template_code = "SMS_123456789"

2. 邀请系统

支持用户邀请功能，注册时可以指定邀请人：

{
    "phone": "13800138000",
    "sms_code": "123456",
    "referer_id": 123  // 邀请人用户ID
}

3. 会员到期提醒

from svc_user_auth_zxw.tools.membership_cleanup import get_expiring_memberships

# 获取即将过期的会员（7天内）
expiring_memberships = await get_expiring_memberships(days=7)

# 发送到期提醒邮件
for membership in expiring_memberships:
    await send_expiry_reminder(membership.user.email, membership)

⚠️ 错误处理

1. 统一错误格式

{
    "code": 400,
    "data": "具体错误信息描述"
}

2. 常见错误代码

HTTP状态码

• 200: 成功
• 400: 请求参数错误
• 401: 未授权（Token无效或过期）
• 403: 权限不足
• 404: 资源未找到
• 500: 服务器内部错误

业务错误代码

• 认证相关: 用户名已注册、无效的用户名或密码、token验证失败
• 验证码相关: 验证码发送失败、验证码验证失败、验证码已过期
• 权限相关: 角色未找到、权限不足、应用未找到
• 会员相关: 会员类型不存在、会员已过期、会员权限不足

3. 前端错误处理示例

// Axios 响应拦截器
axios.interceptors.response.use(
    response => {
        // 成功响应
        if (response.data.code === 200) {
            return response.data.data;
        }
        // 业务错误
        throw new Error(response.data.data);
    },
    error => {
        // HTTP错误处理
        if (error.response?.status === 401) {
            // Token过期，跳转登录
            localStorage.clear();
            window.location.href = '/login';
        } else if (error.response?.status === 403) {
            // 权限不足
            ElMessage.error('权限不足，请联系管理员');
        } else if (error.response?.status === 400) {
            // 业务逻辑错误
            ElMessage.error(error.response?.data?.data || '请求参数错误');
        } else if (error.response?.status === 500) {
            // 服务器错误
            ElMessage.error('服务器内部错误，请稍后重试');
        }
        return Promise.reject(error);
    }
);

🚀 部署指南

1. 环境变量配置

# 必需环境变量
export USER_CENTER_DATABASE_URL="postgresql+asyncpg://user:pass@host:port/db"
export REDIS_URL_AUTH="redis://:password@host:port/db"

# 可选环境变量
export WECHAT_APP_ID="your_wechat_app_id"
export WECHAT_APP_SECRET="your_wechat_app_secret"
export ALIYUN_ACCESS_KEY_ID="your_access_key_id"
export ALIYUN_ACCESS_KEY_SECRET="your_access_key_secret"

2. 生产环境配置

# 生产环境配置示例
class JWT:
    SECRET_KEY = os.environ.get("JWT_SECRET_KEY", "your-production-secret-key")
    ALGORITHM = "HS256"
    expire_time = timedelta(hours=1)  # 生产环境建议较短的过期时间

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-domain.com"],  # 生产环境指定具体域名
    allow_credentials=True,
    allow_methods=["GET", "POST", "DELETE", "PUT"],
    allow_headers=["*"],
)

3. 数据库迁移

# 安装Alembic
pip install alembic

# 初始化迁移环境
alembic init alembic

# 生成迁移文件
alembic revision --autogenerate -m "Initial migration"

# 执行迁移
alembic upgrade head

4. Docker部署示例

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

5. 性能优化建议

数据库索引优化

-- 用户表索引
CREATE INDEX idx_users_username ON users(username);
CREATE INDEX idx_users_phone ON users(phone);
CREATE INDEX idx_users_email ON users(email);

-- 会员表索引
CREATE INDEX idx_memberships_user_id ON memberships(user_id);
CREATE INDEX idx_memberships_end_time ON memberships(end_time);
CREATE INDEX idx_memberships_status ON memberships(status);

-- 角色表索引
CREATE INDEX idx_user_roles_user_id ON user_roles(user_id);
CREATE INDEX idx_user_roles_role_id ON user_roles(role_id);

Redis缓存配置

# Redis连接池配置
REDIS_CONFIG = {
    "host": "localhost",
    "port": 6379,
    "db": 0,
    "password": "your_password",
    "max_connections": 20,
    "retry_on_timeout": True,
    "socket_timeout": 5,
    "socket_connect_timeout": 5
}

6. 监控和日志

应用监控

import logging
from svc_user_auth_zxw.tools import logger

# 配置日志级别
logging.basicConfig(level=logging.INFO)

# 自定义日志处理
@app.middleware("http")
async def log_requests(request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    
    logger.info(
        f"{request.method} {request.url} - "
        f"Status: {response.status_code} - "
        f"Time: {process_time:.3f}s"
    )
    return response

健康检查

@app.get("/health")
async def health_check():
    # 检查数据库连接
    try:
        # 执行简单查询
        await db.execute("SELECT 1")
        db_status = "healthy"
    except Exception as e:
        db_status = f"unhealthy: {str(e)}"
    
    # 检查Redis连接
    try:
        await redis.ping()
        redis_status = "healthy"
    except Exception as e:
        redis_status = f"unhealthy: {str(e)}"
    
    return {
        "status": "ok",
        "database": db_status,
        "redis": redis_status,
        "version": "2.1.0"
    }

📁 示例项目

完整的使用示例请参考项目中的 useExamples 目录：

• main.py: FastAPI 应用集成示例
• api_权限管理.py: 高级权限管理示例
• 会员功能使用示例.py: 会员体系完整使用示例

会员功能详细说明请查看：

• 会员功能说明.md: 会员体系详细功能说明和使用文档

👨‍💻 技术支持

• 作者: 张薛伟 (Zhang Xuewei)
• 邮箱: [e-mail]
• GitHub: https://github.com/sunshineinwater/

如有问题或建议，欢迎提交Issue或联系作者。

�� 许可证

MIT License