使用 Python FastAPI 与 Tortoise ORM 实现账号密码登录和手机验证码登录

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 redis2. 数据库模型设计使用 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(pkTrue) username fields.CharField(max_length50, uniqueTrue, nullTrue) email fields.CharField(max_length100, uniqueTrue, nullTrue) phone fields.CharField(max_length20, uniqueTrue, nullTrue) # 密码相关字段 hashed_password fields.CharField(max_length255, nullTrue) 用户信息 full_name fields.CharField(max_length100, nullTrue) is_active fields.BooleanField(defaultTrue) created_at fields.DatetimeField(auto_now_addTrue) updated_at fields.DatetimeField(auto_nowTrue) class Meta: table users def str(self): return fUser(id{self.id}, username{self.username}) 创建 Pydantic 模型用于序列化/验证 User_Pydantic pydantic_model_creator(User, nameUser) UserIn_Pydantic pydantic_model_creator(User, nameUserIn, exclude_readonlyTrue)2.2 验证码记录模型class VerificationCode(models.Model): 手机验证码记录 id fields.IntField(pkTrue) phone fields.CharField(max_length20, indexTrue) code fields.CharField(max_length6) created_at fields.DatetimeField(auto_now_addTrue) expires_at fields.DatetimeField() is_used fields.BooleanField(defaultFalse) class Meta: table verification_codes def str(self): return fVerificationCode(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], deprecatedauto) 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(minutessettings.access_token_expire_minutes) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, settings.secret_key, algorithmsettings.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_modelTokenResponse) async def password_login(login_data: PasswordLoginRequest): 账号密码登录 支持用户名、邮箱或手机号登录 user None 根据提供的标识查找用户 if login_data.username: try: user await User.get(usernamelogin_data.username) except DoesNotExist: pass elif login_data.email: try: user await User.get(emaillogin_data.email) except DoesNotExist: pass elif login_data.phone: try: user await User.get(phonelogin_data.phone) except DoesNotExist: pass 用户不存在或密码错误 if not user or not user.hashed_password: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail用户名或密码错误 ) 验证密码 if not verify_password(login_data.password, user.hashed_password): raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail用户名或密码错误 ) 检查用户是否激活 if not user.is_active: raise HTTPException( status_codestatus.HTTP_403_FORBIDDEN, detail用户账户已被禁用 ) 创建访问令牌 access_token create_access_token( data{sub: str(user.id), username: user.username} ) return TokenResponse( access_tokenaccess_token, user_iduser.id, usernameuser.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_responsesTrue) async def generate_verification_code(phone: str) - str: 生成并存储验证码 生成6位数字验证码 code str(random.randint(100000, 999999)) 存储到数据库 expires_at datetime.utcnow() timedelta(minutes5) await VerificationCode.create( phonephone, codecode, expires_atexpires_at ) 同时缓存到 Redis (可选提升性能) redis_key fverify_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 fverify_code:{phone} cached_code redis_client.get(redis_key) if cached_code and cached_code code: # 验证成功清除缓存 redis_client.delete(redis_key) # 标记数据库中的验证码为已使用 await VerificationCode.filter(phonephone, codecode, is_usedFalse).update(is_usedTrue) return True 如果缓存没有检查数据库 now datetime.utcnow() verification await VerificationCode.filter( phonephone, codecode, is_usedFalse, expires_at__gtnow ).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_codestatus.HTTP_400_BAD_REQUEST, detail手机号格式不正确 ) 检查用户是否存在如果不存在则创建 user, created await User.get_or_create( phonerequest.phone, defaults{ username: fuser_{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_modelTokenResponse) async def phone_login(login_data: PhoneLoginRequest): 手机验证码登录 验证验证码 if not await verify_phone_code(login_data.phone, login_data.code): raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail验证码错误或已过期 ) 获取或创建用户 user await User.get_or_none(phonelogin_data.phone) if not user: # 如果用户不存在创建新用户 user await User.create( phonelogin_data.phone, usernamefuser_{login_data.phone}, is_activeTrue ) 检查用户是否激活 if not user.is_active: raise HTTPException( status_codestatus.HTTP_403_FORBIDDEN, detail用户账户已被禁用 ) 创建访问令牌 access_token create_access_token( data{sub: str(user.id), phone: user.phone} ) return TokenResponse( access_tokenaccess_token, user_iduser.id, usernameuser.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_codestatus.HTTP_201_CREATED) async def register_user(user_data: UserRegisterRequest): 用户注册 检查用户名是否已存在 if await User.exists(usernameuser_data.username): raise HTTPException( status_codestatus.HTTP_400_BAD_REQUEST, detail用户名已存在 ) 检查邮箱是否已存在 if await User.exists(emailuser_data.email): raise HTTPException( status_codestatus.HTTP_400_BAD_REQUEST, detail邮箱已存在 ) 检查手机号是否已存在 if await User.exists(phoneuser_data.phone): raise HTTPException( status_codestatus.HTTP_400_BAD_REQUEST, detail手机号已存在 ) 创建用户 hashed_password get_password_hash(user_data.password) user await User.create( usernameuser_data.username, emailuser_data.email, phoneuser_data.phone, hashed_passwordhashed_password, full_nameuser_data.full_name, is_activeTrue ) 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_urlsettings.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( titleFastAPI 认证系统, description使用 Tortoise ORM 实现账号密码和手机验证码登录, version1.0.0, lifespanlifespan ) 配置 CORS app.add_middleware( CORSMiddleware, allow_origins[], # 生产环境应限制具体域名 allow_credentialsTrue, 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, host0.0.0.0, port8000) 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.pyFastAPI 应用主入口包含应用配置、中间件和路由注册 config.py应用配置文件使用 Pydantic 管理环境变量 models.py数据库模型定义包含用户和验证码模型 schemas.pyPydantic 模型定义用于请求/响应数据验证 auth.py认证相关工具函数包括密码哈希和 JWT 处理 routers.pyAPI 路由定义包含登录、注册等接口 database.py数据库初始化和连接管理 requirements.txt项目依赖包列表 .env环境变量配置文件