当前位置： 首页 > news >正文

FastAPI 小白教程：从入门级到实战(源码教程)

news 2025/7/5 11:07:06

目录

1. FastAPI 基本介绍

安装 FastAPI

2. 简单的 CRUD 示例

2.1 创建基本应用

2.2 添加 CRUD 操作​​​​​​​

3. 处理跨域请求 (CORS)

4. 普通案例：待办事项 API​​​​​​​

5. 企业案例：认证和数据库集成

5.1 使用 SQLAlchemy 和 JWT 认证​​​​​​​

6. 常见问题

6.1 性能问题

6.2 异步支持

6.3 数据库迁移

6.4 部署问题

6.5 测试

6.6 文件上传

 

1. FastAPI 基本介绍

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于 Python 构建 API。它具有以下特点：

  • 快速：与 NodeJS 和 Go 相当的高性能

  • 快速编码：开发功能的速度提高约 200%-300%

  • 更少的错误：减少约 40% 的人为错误

  • 直观：强大的编辑器支持

  • 简单：易于使用和学习

  • 简短：最小化代码重复

  • 稳健：生产就绪的代码

  • 基于标准：完全兼容 API 开放标准（如 OpenAPI 和 JSON Schema）

安装 FastAPI

pip install fastapipip install uvicorn  # ASGI 服务器

2. 简单的 CRUD 示例

让我们创建一个简单的用户管理 API，包含创建、读取、更新和删除操作。

2.1 创建基本应用​​​​​​​

from fastapi import FastAPIapp = FastAPI()# 模拟数据库fake_db = []@app.get("/")def read_root():    return {"message": "Welcome to FastAPI!"}

运行应用：

uvicorn main:app --reload

2.2 添加 CRUD 操作​​​​​​​

from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom typing import Listapp = FastAPI()# 用户模型class User(BaseModel):    id: int    name: str    email: str# 模拟数据库fake_db: List[User] = []# 创建用户@app.post("/users/", response_model=User)def create_user(user: User):    fake_db.append(user)    return user# 获取所有用户@app.get("/users/", response_model=List[User])def read_users():    return fake_db# 获取单个用户@app.get("/users/{user_id}", response_model=User)def read_user(user_id: int):    for user in fake_db:        if user.id == user_id:            return user    raise HTTPException(status_code=404, detail="User not found")# 更新用户@app.put("/users/{user_id}", response_model=User)def update_user(user_id: int, updated_user: User):    for index, user in enumerate(fake_db):        if user.id == user_id:            fake_db[index] = updated_user            return updated_user    raise HTTPException(status_code=404, detail="User not found")# 删除用户@app.delete("/users/{user_id}")def delete_user(user_id: int):    for index, user in enumerate(fake_db):        if user.id == user_id:            del fake_db[index]            return {"message": "User deleted"}    raise HTTPException(status_code=404, detail="User not found")

3. 处理跨域请求 (CORS)

当你的前端应用运行在不同的域名下时，需要处理跨域请求：​​​​​​​

from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()# 设置允许的源origins = [    "http://localhost",    "http://localhost:8080",    "http://localhost:3000",]# 添加中间件app.add_middleware(    CORSMiddleware,    allow_origins=origins,    allow_credentials=True,    allow_methods=["*"],    allow_headers=["*"],)

4. 普通案例：待办事项 API​​​​​​​

from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom typing import List, Optionalapp = FastAPI()class Todo(BaseModel):    id: int    title: str    description: Optional[str] = None    completed: bool = Falsetodos = []@app.post("/todos/", response_model=Todo)def create_todo(todo: Todo):    todos.append(todo)    return todo@app.get("/todos/", response_model=List[Todo])def read_todos(completed: Optional[bool] = None):    if completed is None:        return todos    return [todo for todo in todos if todo.completed == completed]@app.put("/todos/{todo_id}", response_model=Todo)def update_todo(todo_id: int, updated_todo: Todo):    for index, todo in enumerate(todos):        if todo.id == todo_id:            todos[index] = updated_todo            return updated_todo    raise HTTPException(status_code=404, detail="Todo not found")@app.delete("/todos/{todo_id}")def delete_todo(todo_id: int):    for index, todo in enumerate(todos):        if todo.id == todo_id:            del todos[index]            return {"message": "Todo deleted"}    raise HTTPException(status_code=404, detail="Todo not found")

5. 企业案例：认证和数据库集成

5.1 使用 SQLAlchemy 和 JWT 认证​​​​​​​

from fastapi import FastAPI, Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestFormfrom pydantic import BaseModelfrom datetime import datetime, timedeltafrom jose import JWTError, jwtfrom passlib.context import CryptContextfrom typing import Optionalfrom sqlalchemy import create_engine, Column, Integer, String, Booleanfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker, Session# 数据库配置SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"engine = create_engine(SQLALCHEMY_DATABASE_URL)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)Base = declarative_base()# 数据库模型class DBUser(Base):    __tablename__ = "users"    id = Column(Integer, primary_key=True, index=True)    username = Column(String, unique=True, index=True)    hashed_password = Column(String)    is_active = Column(Boolean, default=True)Base.metadata.create_all(bind=engine)# JWT 配置SECRET_KEY = "your-secret-key"ALGORITHM = "HS256"ACCESS_TOKEN_EXPIRE_MINUTES = 30class Token(BaseModel):    access_token: str    token_type: strclass TokenData(BaseModel):    username: Optional[str] = Noneclass User(BaseModel):    username: str    is_active: Optional[bool] = Noneclass UserInDB(User):    hashed_password: strpwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")app = FastAPI()# 数据库依赖def get_db():    db = SessionLocal()    try:        yield db    finally:        db.close()# 密码哈希def verify_password(plain_password, hashed_password):    return pwd_context.verify(plain_password, hashed_password)def get_password_hash(password):    return pwd_context.hash(password)# 用户认证def authenticate_user(db: Session, username: str, password: str):    user = db.query(DBUser).filter(DBUser.username == username).first()    if not user:        return False    if not verify_password(password, user.hashed_password):        return False    return user# 创建 tokendef create_access_token(data: dict, expires_delta: Optional[timedelta] = None):    to_encode = data.copy()    if expires_delta:        expire = datetime.utcnow() + expires_delta    else:        expire = datetime.utcnow() + timedelta(minutes=15)    to_encode.update({"exp": expire})    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)    return encoded_jwt# 获取当前用户async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):    credentials_exception = HTTPException(        status_code=status.HTTP_401_UNAUTHORIZED,        detail="Could not validate credentials",        headers={"WWW-Authenticate": "Bearer"},    )    try:        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])        username: str = payload.get("sub")        if username is None:            raise credentials_exception        token_data = TokenData(username=username)    except JWTError:        raise credentials_exception    user = db.query(DBUser).filter(DBUser.username == token_data.username).first()    if user is None:        raise credentials_exception    return user# 路由@app.post("/token", response_model=Token)async def login_for_access_token(    form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):    user = authenticate_user(db, form_data.username, form_data.password)    if not user:        raise HTTPException(            status_code=status.HTTP_401_UNAUTHORIZED,            detail="Incorrect username or password",            headers={"WWW-Authenticate": "Bearer"},        )    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)    access_token = create_access_token(        data={"sub": user.username}, expires_delta=access_token_expires    )    return {"access_token": access_token, "token_type": "bearer"}@app.get("/users/me/", response_model=User)async def read_users_me(current_user: User = Depends(get_current_user)):    return current_user@app.post("/users/", response_model=User)def create_user(user: User, db: Session = Depends(get_db)):    hashed_password = get_password_hash(user.password)    db_user = DBUser(username=user.username, hashed_password=hashed_password)    db.add(db_user)    db.commit()    db.refresh(db_user)    return db_user

6. 常见问题

6.1 性能问题

Q: FastAPI 真的比 Flask 快吗？

A: 是的，FastAPI 基于 Starlette（用于 ASGI）和 Pydantic，性能接近 NodeJS 和 Go。根据基准测试，FastAPI 比 Flask 快得多，特别是在处理大量请求时。

6.2 异步支持

Q: 如何在 FastAPI 中正确使用 async/await？

A: FastAPI 完全支持异步。对于 I/O 密集型操作（如数据库查询、API 调用），使用 async def：​​​​​​​

@app.get("/items/{item_id}")async def read_item(item_id: int):    item = await some_async_function(item_id)    return item

6.3 数据库迁移

Q: 如何管理数据库迁移？

A: 推荐使用 Alembic：​​​​​​​

pip install alembicalembic init migrations

然后配置 alembic.ini 和 migrations/env.py 文件。

6.4 部署问题

Q: 如何部署 FastAPI 应用？

A: 常见部署方式：

  1. 使用 Uvicorn 或 Hypercorn 作为 ASGI 服务器

  2. 使用 Gunicorn 作为进程管理器（配合 Uvicorn worker）

  3. 使用 Docker 容器化

  4. 部署到云平台（如 AWS, GCP, Azure）

6.5 测试

Q: 如何测试 FastAPI 应用？

A: 使用 TestClient：​​​​​​​

from fastapi.testclient import TestClientclient = TestClient(app)def test_read_main():    response = client.get("/")    assert response.status_code == 200    assert response.json() == {"message": "Welcome to FastAPI!"}

6.6 文件上传

Q: 如何处理文件上传？

A: 使用 File 和 UploadFile：​​​​​​​

from fastapi import FastAPI, File, UploadFile@app.post("/uploadfile/")async def create_upload_file(file: UploadFile = File(...)):    return {"filename": file.filename}

FastAPI 是一个强大而现代的 Python Web 框架，适合构建高性能的 API。它结合了 Python 类型提示、自动文档生成和异步支持，使得开发体验非常愉快。通过本教程，你应该已经掌握了 FastAPI 的基本用法、CRUD 操作、跨域处理、认证授权等核心功能。

* Thanks you *

如果觉得文章内容不错，随手帮忙点个赞、在看、转发一下，如果想第一时间收到推送，也可以给我个星标⭐～谢谢你看我的文章。

*往期推荐 *

实现如何利用 Kafka 延时删除 用户邮箱的验证码（如何发送邮箱+源码) - 第一期

Docker小白入门教程一篇领你入门(CRUD全命令+无废话版+问题集)-第三期

Docker小白入门教程一篇领你入门(CRUD全命令+无废话版+问题集)

想要高效处理，那不妨看看 Python的 异步 Asyncio 保证效率翻多倍

银河麒麟 | ubuntu 安装国产达梦DM8数据库(安装+外网通+IDEA连接)

网络设备日志存储到指定的Kiwi-log服务器(图解+软件)

银河麒麟 | ubuntu 安装运用 docker 容器，实现容器化部署项目

银河麒麟 | ubuntu 安装zabbix监控设备信息(亲测包对)

国产操作系统－银河麒麟本地化部署Ollama国产开源的AI大模型Qwen3

Ubuntu |  安装 Zabbix 一篇就够了

Swagger | 手把手带你写自动生成接口文档的爽感(零基础亲测实用)

SpringBoot整合Openfeign接入Kimi Ai!!超简单，居然没多少行代码？？(附加兜底教程)

SpringBoot接入Kimi实践记录轻松上手

Linux | 零基础Ubuntu搭建JDK

Maven | 站在初学者的角度配置与项目创建(新手必学会)

Spring Ai | 极简代码从零带你一起走进AI项目(中英)

Open Ai | 从零搭建属于你的Ai项目(中英结合)

MongoDB | 零基础学习与Springboot整合ODM实现增删改查(附源码)

Openfeign | 只传递城市代码，即可获取该地域实时的天气数据(免费的天气API)

Redis | 缓存技术对后端的重要性，你知道多少？

Mongodb | 基于Springboot开发综合社交网络应用的项目案例(中英)

感谢阅读 | 更多内容尽在公棕号 WMCode | CSDN@小Mie不吃饭

http://www.dtcms.com/a/266699.html

相关文章：

  • springboot整合腾讯云cos对象存储，获取临时密钥，前端直传图片文件
  • Spring Cloud网关与CI文件配置请求安全性对比
  • 基于二维码的视频合集高效管理与分发技术
  • monorepo + Turborepo --- 运行任务
  • MySQL ON DUPLICATE KEY UPDATE 用法详解
  • 鸿蒙开发List长按Item拖拽切换效果
  • 基于区块链的物联网（IoT）安全通信与数据共享的典型实例
  • JSONLines和JSON数据格式使用教程
  • AI大模型：（二）1.5 Stable Diffusion中文文生图模型部署
  • 30 秒锁定黑客攻击：SLS SQL 如何从海量乱序日志中“揪”出攻击源
  • 【C语言刷题】第十天：加量加餐继续，代码题训练，融会贯通IO模式
  • 短篇小说7.4
  • BM4 合并两个排序的链表
  • QT6 源（152）模型视图架构里的表格窗体视图 QTableWidget 篇二：学习本类的 protected 权限的成员函数，以及信号与槽函数
  • c语言中的函数IV
  • MCMC：高维概率采样的“随机游走”艺术
  • pybind11 导出 C++ map 在 Python 层 get 访问慢的优化方案
  • 区块链技术核心组件及应用架构的全面解析
  • python打卡day59@浙大疏锦行
  • 车载电子电气架构 --- OEM走向开放协同与敏捷迭代
  • 数据结构：队列的顺序存储实现
  • 【Linux 系统】基础IO——Linux中对文件的理解
  • 【深度学习新浪潮】如何使用大模型等技术基于序列预测蛋白质的结构，功能和靶点？
  • 【学习笔记】Lean4基础 ing
  • 邮科千兆8光8电工业级交换机互联网的脉搏
  • 洛谷刷题8
  • 云原生Kubernetes系列 | Ingress和Egress网络策略NetworkPolicy结合案例使用详解
  • 5060Ti安装黑屏问题一解
  • 【WIP】【VLAVLM——InternVL系列】
  • Maven编译和打包插件