企业官网建设流程全解析

1. 项目概述与准备工作

本文将详细介绍如何使用 Python 的 FastAPI 框架结合 Tortoise ORM 实现两种常见的用户登录方式：传统的账号密码登录和现代化的手机验证码登录。我们将从环境搭建、数据库设计、核心逻辑实现到完整 API 测试，一步步构建一个安全、可扩展的认证系统。

1.1 技术栈说明

• FastAPI: 现代、高性能的 Python Web 框架，用于构建 API。
• Tortoise ORM: 异步友好的 Python ORM，支持 SQLite、PostgreSQL、MySQL 等数据库。
• Pydantic: 用于数据验证和设置管理。
• JWT (JSON Web Tokens): 用于生成和验证访问令牌。
• Redis(可选): 用于缓存验证码和会话信息，提升性能。

1.2 环境准备

首先创建项目目录并安装必要的依赖：

mkdir fastapi-auth-demo cd fastapi-auth-demo python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install fastapi uvicorn tortoise-orm pydantic python-jose[cryptography] passlib[bcrypt] python-multipart redis

2. 数据库模型设计

使用 Tortoise ORM 定义用户表，同时支持密码和手机号两种认证方式。

2.1 用户模型 (models.py)

from tortoise import fields, models from tortoise.contrib.pydantic import pydantic_model_creator class User(models.Model): """用户模型""" id = fields.IntField(pk=True) username = fields.CharField(max_length=50, unique=True, null=True) email = fields.CharField(max_length=100, unique=True, null=True) phone = fields.CharField(max_length=20, unique=True, null=True) # 密码相关字段 hashed_password = fields.CharField(max_length=255, null=True) 用户信息 full_name = fields.CharField(max_length=100, null=True) is_active = fields.BooleanField(default=True) created_at = fields.DatetimeField(auto_now_add=True) updated_at = fields.DatetimeField(auto_now=True) class Meta: table = "users" def str(self): return f"User(id={self.id}, username={self.username})" 创建 Pydantic 模型用于序列化/验证 User_Pydantic = pydantic_model_creator(User, name="User") UserIn_Pydantic = pydantic_model_creator(User, name="UserIn", exclude_readonly=True)

2.2 验证码记录模型

class VerificationCode(models.Model): """手机验证码记录""" id = fields.IntField(pk=True) phone = fields.CharField(max_length=20, index=True) code = fields.CharField(max_length=6) created_at = fields.DatetimeField(auto_now_add=True) expires_at = fields.DatetimeField() is_used = fields.BooleanField(default=False) class Meta: table = "verification_codes" def str(self): return f"VerificationCode(phone={self.phone}, code={self.code})"</code></pre> 3. 核心配置与工具函数 3.1 配置文件 (config.py) from pydantic import BaseSettings class Settings(BaseSettings): """应用配置""" app_name: str = "FastAPI Auth Demo" debug: bool = True JWT 配置 secret_key: str = "your-secret-key-change-in-production" algorithm: str = "HS256" access_token_expire_minutes: int = 30 数据库配置 database_url: str = "sqlite://db.sqlite3" Redis 配置 (用于缓存验证码) redis_url: str = "redis://localhost:6379/0" 短信服务配置 (示例) sms_api_key: str = "your-sms-api-key" class Config: env_file = ".env" settings = Settings() 3.2 密码哈希工具 (auth.py) from passlib.context import CryptContext 密码哈希上下文 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def verify_password(plain_password: str, hashed_password: str) -> bool: """验证密码""" return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: """生成密码哈希""" return pwd_context.hash(password) 3.3 JWT 工具函数 from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from config import settings def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): """创建 JWT 访问令牌""" to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.access_token_expire_minutes) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm) return encoded_jwt def verify_token(token: str): """验证 JWT 令牌""" try: payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm]) return payload except JWTError: return None 4. 账号密码登录实现 4.1 请求/响应模型 from pydantic import BaseModel, EmailStr from typing import Optional class PasswordLoginRequest(BaseModel): """密码登录请求""" username: Optional[str] = None email: Optional[EmailStr] = None phone: Optional[str] = None password: str class Config: schema_extra = { "example": { "username": "john_doe", "password": "securepassword123" } } class TokenResponse(BaseModel): """令牌响应""" access_token: str token_type: str = "bearer" user_id: int username: Optional[str] = None 4.2 密码登录 API from fastapi import APIRouter, Depends, HTTPException, status from tortoise.exceptions import DoesNotExist from models import User from schemas import PasswordLoginRequest, TokenResponse from auth import verify_password, create_access_token router = APIRouter(prefix="/auth", tags=["authentication"]) @router.post("/login/password", response_model=TokenResponse) async def password_login(login_data: PasswordLoginRequest): """ 账号密码登录 支持用户名、邮箱或手机号登录 """ user = None 根据提供的标识查找用户 if login_data.username: try: user = await User.get(username=login_data.username) except DoesNotExist: pass elif login_data.email: try: user = await User.get(email=login_data.email) except DoesNotExist: pass elif login_data.phone: try: user = await User.get(phone=login_data.phone) except DoesNotExist: pass 用户不存在或密码错误 if not user or not user.hashed_password: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误" ) 验证密码 if not verify_password(login_data.password, user.hashed_password): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误" ) 检查用户是否激活 if not user.is_active: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="用户账户已被禁用" ) 创建访问令牌 access_token = create_access_token( data={"sub": str(user.id), "username": user.username} ) return TokenResponse( access_token=access_token, user_id=user.id, username=user.username )</code></pre> 5. 手机验证码登录实现 5.1 验证码发送与验证 import random import redis from datetime import datetime, timedelta from fastapi import HTTPException, status from models import VerificationCode from config import settings 初始化 Redis 客户端 redis_client = redis.from_url(settings.redis_url, decode_responses=True) async def generate_verification_code(phone: str) -> str: """生成并存储验证码""" 生成6位数字验证码 code = str(random.randint(100000, 999999)) 存储到数据库 expires_at = datetime.utcnow() + timedelta(minutes=5) await VerificationCode.create( phone=phone, code=code, expires_at=expires_at ) 同时缓存到 Redis (可选，提升性能) redis_key = f"verify_code:{phone}" redis_client.setex(redis_key, 300, code) # 5分钟过期 这里应该调用短信服务发送验证码 await send_sms(phone, f"您的验证码是：{code}，5分钟内有效") return code async def verify_phone_code(phone: str, code: str) -> bool: """验证手机验证码""" 先检查 Redis 缓存 redis_key = f"verify_code:{phone}" cached_code = redis_client.get(redis_key) if cached_code and cached_code == code: # 验证成功，清除缓存 redis_client.delete(redis_key) # 标记数据库中的验证码为已使用 await VerificationCode.filter(phone=phone, code=code, is_used=False).update(is_used=True) return True 如果缓存没有，检查数据库 now = datetime.utcnow() verification = await VerificationCode.filter( phone=phone, code=code, is_used=False, expires_at__gt=now ).first() if verification: # 验证成功 verification.is_used = True await verification.save() # 清除可能的缓存 redis_client.delete(redis_key) return True return False</code></pre> 5.2 手机登录 API from pydantic import BaseModel from fastapi import APIRouter, HTTPException, status class PhoneLoginRequest(BaseModel): """手机登录请求""" phone: str code: str class SendCodeRequest(BaseModel): """发送验证码请求""" phone: str @router.post("/login/phone/send-code") async def send_verification_code(request: SendCodeRequest): """发送手机验证码""" 验证手机号格式 (这里简单示例，实际需要更严格的验证) if not request.phone.startswith("1") or len(request.phone) != 11: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="手机号格式不正确" ) 检查用户是否存在，如果不存在则创建 user, created = await User.get_or_create( phone=request.phone, defaults={ "username": f"user_{request.phone}", "is_active": True } ) 生成并发送验证码 code = await generate_verification_code(request.phone) 在实际项目中，这里应该调用短信服务 为了演示，我们返回验证码（生产环境不要这样做！） return { "message": "验证码已发送", "phone": request.phone, "code": code, # 仅用于演示，生产环境移除 "user_created": created } @router.post("/login/phone", response_model=TokenResponse) async def phone_login(login_data: PhoneLoginRequest): """手机验证码登录""" 验证验证码 if not await verify_phone_code(login_data.phone, login_data.code): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="验证码错误或已过期" ) 获取或创建用户 user = await User.get_or_none(phone=login_data.phone) if not user: # 如果用户不存在，创建新用户 user = await User.create( phone=login_data.phone, username=f"user_{login_data.phone}", is_active=True ) 检查用户是否激活 if not user.is_active: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="用户账户已被禁用" ) 创建访问令牌 access_token = create_access_token( data={"sub": str(user.id), "phone": user.phone} ) return TokenResponse( access_token=access_token, user_id=user.id, username=user.username )</code></pre> 6. 用户注册与数据库初始化 6.1 用户注册 API from pydantic import BaseModel, EmailStr from fastapi import HTTPException, status from auth import get_password_hash class UserRegisterRequest(BaseModel): """用户注册请求""" username: str email: EmailStr phone: str password: str full_name: Optional[str] = None @router.post("/register", status_code=status.HTTP_201_CREATED) async def register_user(user_data: UserRegisterRequest): """用户注册""" 检查用户名是否已存在 if await User.exists(username=user_data.username): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="用户名已存在" ) 检查邮箱是否已存在 if await User.exists(email=user_data.email): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="邮箱已存在" ) 检查手机号是否已存在 if await User.exists(phone=user_data.phone): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="手机号已存在" ) 创建用户 hashed_password = get_password_hash(user_data.password) user = await User.create( username=user_data.username, email=user_data.email, phone=user_data.phone, hashed_password=hashed_password, full_name=user_data.full_name, is_active=True ) return { "message": "用户注册成功", "user_id": user.id, "username": user.username }</code></pre> 6.2 数据库初始化 from tortoise import Tortoise from config import settings async def init_db(): """初始化数据库""" await Tortoise.init( db_url=settings.database_url, modules={"models": ["models"]} ) await Tortoise.generate_schemas() print("数据库初始化完成") async def close_db(): """关闭数据库连接""" await Tortoise.close_connections() 7. 主应用与路由集成 7.1 主应用文件 (main.py) from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager from routers import auth_router from database import init_db, close_db @asynccontextmanager async def lifespan(app: FastAPI): """应用生命周期管理""" 启动时初始化数据库 await init_db() yield 关闭时清理资源 await close_db() app = FastAPI( title="FastAPI 认证系统", description="使用 Tortoise ORM 实现账号密码和手机验证码登录", version="1.0.0", lifespan=lifespan ) 配置 CORS app.add_middleware( CORSMiddleware, allow_origins=[""], # 生产环境应限制具体域名 allow_credentials=True, allow_methods=[""], allow_headers=["*"], ) 注册路由 app.include_router(auth_router) @app.get("/") async def root(): return {"message": "FastAPI 认证系统已启动"} if name == "main": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) 7.2 项目结构 以下是完整的项目目录结构： fastapi-auth-demo/ ├── main.py # 主应用入口 ├── config.py # 配置文件 ├── models.py # 数据库模型 ├── schemas.py # Pydantic 模型 ├── auth.py # 认证工具函数 ├── routers.py # API 路由 ├── database.py # 数据库初始化 ├── requirements.txt # 依赖文件 └── .env # 环境变量文件 项目结构说明： main.py：FastAPI 应用主入口，包含应用配置、中间件和路由注册 config.py：应用配置文件，使用 Pydantic 管理环境变量 models.py：数据库模型定义，包含用户和验证码模型 schemas.py：Pydantic 模型定义，用于请求/响应数据验证 auth.py：认证相关工具函数，包括密码哈希和 JWT 处理 routers.py：API 路由定义，包含登录、注册等接口 database.py：数据库初始化和连接管理 requirements.txt：项目依赖包列表 .env：环境变量配置文件