在后端开发中RESTful API 是最常见的服务形式之一。本文将使用 Python 和 FastAPI从零构建一个简单的用户管理系统实现用户的新增、查询、修改和删除功能。本文示例代码结构清晰、依赖较少适合初学者学习也可以作为小型项目的基础模板。一、技术栈Python 3.10FastAPIUvicornSQLitePydanticFastAPI 的特点是开发速度快自动生成接口文档支持异步编程参数校验方便性能优秀二、创建项目创建项目目录mkdir fastapi-user-demo cd fastapi-user-demo创建虚拟环境python -m venv venvWindows 系统激活虚拟环境venv\Scripts\activateLinux 或 macOS 系统激活虚拟环境source venv/bin/activate安装依赖pip install fastapi uvicorn也可以创建requirements.txtfastapi0.115.0 uvicorn0.30.6然后执行pip install -r requirements.txt三、项目结构项目结构如下fastapi-user-demo/ ├── main.py └── users.db其中main.py接口主程序users.dbSQLite 数据库文件首次启动时自动创建四、编写接口代码创建main.py文件import sqlite3 from contextlib import closing from typing import Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field app FastAPI( title用户管理 API, description一个基于 FastAPI 和 SQLite 的简单用户管理服务, version1.0.0 ) DATABASE users.db class UserCreate(BaseModel): name: str Field(..., min_length1, max_length50) email: str Field(..., min_length5, max_length100) age: int Field(..., ge1, le120) class UserUpdate(BaseModel): name: Optional[str] Field(None, min_length1, max_length50) email: Optional[str] Field(None, min_length5, max_length100) age: Optional[int] Field(None, ge1, le120) class User(BaseModel): id: int name: str email: str age: int def get_connection(): connection sqlite3.connect(DATABASE) connection.row_factory sqlite3.Row return connection def init_database(): with closing(get_connection()) as connection: connection.execute( CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, email TEXT NOT NULL UNIQUE, age INTEGER NOT NULL ) ) connection.commit() app.on_event(startup) def startup_event(): init_database() app.get(/) def index(): return { message: 用户管理 API 已启动 } app.post(/users, response_modelUser, status_code201) def create_user(user: UserCreate): try: with closing(get_connection()) as connection: cursor connection.execute( INSERT INTO users (name, email, age) VALUES (?, ?, ?) , (user.name, user.email, user.age) ) connection.commit() user_id cursor.lastrowid row connection.execute( SELECT * FROM users WHERE id ?, (user_id,) ).fetchone() return dict(row) except sqlite3.IntegrityError: raise HTTPException( status_code400, detail邮箱已存在不能重复注册 ) app.get(/users, response_modellist[User]) def get_users(): with closing(get_connection()) as connection: rows connection.execute( SELECT * FROM users ORDER BY id DESC ).fetchall() return [dict(row) for row in rows] app.get(/users/{user_id}, response_modelUser) def get_user(user_id: int): with closing(get_connection()) as connection: row connection.execute( SELECT * FROM users WHERE id ?, (user_id,) ).fetchone() if row is None: raise HTTPException( status_code404, detail用户不存在 ) return dict(row) app.put(/users/{user_id}, response_modelUser) def update_user(user_id: int, user: UserUpdate): update_data user.model_dump(exclude_unsetTrue) if not update_data: raise HTTPException( status_code400, detail至少提供一个需要修改的字段 ) fields [] values [] for field, value in update_data.items(): fields.append(f{field} ?) values.append(value) values.append(user_id) with closing(get_connection()) as connection: cursor connection.execute( f UPDATE users SET {, .join(fields)} WHERE id ? , values ) connection.commit() if cursor.rowcount 0: raise HTTPException( status_code404, detail用户不存在 ) row connection.execute( SELECT * FROM users WHERE id ?, (user_id,) ).fetchone() return dict(row) app.delete(/users/{user_id}) def delete_user(user_id: int): with closing(get_connection()) as connection: cursor connection.execute( DELETE FROM users WHERE id ?, (user_id,) ) connection.commit() if cursor.rowcount 0: raise HTTPException( status_code404, detail用户不存在 ) return { message: 用户删除成功 }五、启动项目在项目根目录执行uvicorn main:app --reload启动成功后可以看到类似输出Uvicorn running on http://127.0.0.1:8000访问首页http://127.0.0.1:8000返回结果{ message: 用户管理 API 已启动 }六、使用自动生成的接口文档FastAPI 会自动生成 Swagger 接口文档。浏览器访问http://127.0.0.1:8000/docs还可以访问 ReDoc 文档http://127.0.0.1:8000/redoc在 Swagger 页面中可以直接测试所有接口不需要额外安装 Postman。七、测试新增用户请求地址POST /users请求参数{ name: 张三, email: zhangsanexample.com, age: 25 }返回结果{ id: 1, name: 张三, email: zhangsanexample.com, age: 25 }八、查询用户列表请求地址GET /users返回结果[ { id: 1, name: 张三, email: zhangsanexample.com, age: 25 } ]九、修改用户信息请求地址PUT /users/1请求参数{ age: 26 }返回结果{ id: 1, name: 张三, email: zhangsanexample.com, age: 26 }由于UserUpdate中的字段都是可选的所以修改时只需要提交需要修改的内容即可。十、删除用户请求地址DELETE /users/1返回结果{ message: 用户删除成功 }十一、代码中的几个关键点1. 使用 Pydantic 进行参数校验age: int Field(..., ge1, le120)这行代码表示年龄必须是整数并且范围在 1 到 120 之间。如果传入非法参数FastAPI 会自动返回 422 错误。2. 使用参数化查询防止 SQL 注入代码中使用了connection.execute( SELECT * FROM users WHERE id ?, (user_id,) )不要直接拼接用户输入例如# 不推荐 sql fSELECT * FROM users WHERE id {user_id}参数化查询可以有效降低 SQL 注入风险。3. 使用 response_model 规范返回结构app.get(/users, response_modellist[User])response_model可以限制接口返回字段同时让接口文档更加清晰。4. 使用 HTTP 状态码表达结果本文使用了常见状态码200请求成功201资源创建成功400请求参数错误404资源不存在422参数校验失败合理使用状态码有助于前端和其他服务准确处理接口结果。十二、后续可以如何扩展这个示例只是一个基础版本实际项目中还可以继续增加用户登录和 JWT 鉴权分页查询条件搜索数据库迁移SQLAlchemy ORM日志记录统一异常处理Docker 部署Redis 缓存接口限流单元测试和接口测试如果项目规模逐渐扩大建议将代码拆分为多个模块例如app/ ├── main.py ├── database.py ├── models.py ├── schemas.py └── routers/ └── users.py总结本文使用 FastAPI 和 SQLite 实现了一个完整的用户管理 API涵盖了项目环境搭建数据库初始化新增用户查询用户修改用户删除用户参数校验自动接口文档FastAPI 非常适合构建现代 Python Web 服务。对于个人项目、后台管理系统和微服务接口来说它都是一个值得学习的框架。如果你正在学习 Python 后端开发可以在本文代码的基础上继续加入登录认证、分页和权限管理逐步完善成一个真正可用的项目。