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

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

news 2025/7/5 16:49:35

目录

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.xdnf.cn/news/1081045.html

相关文章：

  • 学习者的Python项目灵感
  • WPF 右键菜单 MenuItem 绑定图片时只显示最后一个 Icon
  • 【python实用小脚本-128】基于 Python 的 Hacker News 爬虫工具：自动化抓取新闻数据
  • 第二章-AIGC入门-开启AIGC音频探索之旅：从入门到实践（6/36）
  • 玩转n8n工作流教程（一）：Windows系统本地部署n8n自动化工作流（n8n中文汉化）
  • 【基础算法】贪心 (二) ：推公式
  • 基于大模型的强直性脊柱炎全周期预测与诊疗方案研究
  • 算法学习笔记：6.深度优先搜索算法——从原理到实战，涵盖 LeetCode 与考研 408 例题
  • 风平浪静、无事发生
  • 八股学习（三）---MySQL
  • 【C语言刷题】第十天：加量加餐继续，代码题训练，融会贯通IO模式
  • 类图+案例+代码详解：软件设计模式----原型模式
  • WPF+HelixToolkit打造炫酷自定义3D贴图立方体盒子模型
  • goole chrome变更默认搜索引擎为百度
  • 篇二 OSI七层模型，TCP/IP四层模型，路由器与交换机原理
  • SpringBoot-规划多模块目录结构
  • 从0开始学习R语言--Day38--辛普森多样性指数
  • Nuxt 3 面试题合集（中高级）
  • vue3 获取选中的el-table行数据
  • 浅谈 webshell 构造之如何获取恶意函数
  • 「AI产业」| 《中国信通院华为:智能体技术和应用研究报告》
  • Response对象
  • Excel 如何处理更复杂的嵌套逻辑判断？
  • 网安系列【4】之OWASP与OWASP Top 10：Web安全入门指南
  • Flink Vitess CDC 环境配置与验证
  • QString 转 varchar
  • 【网络与系统安全】域类实施模型DTE
  • 数字资产革命中的信任之锚：RWA法律架构的隐形密码
  • ORACLE 日常查询
  • 浏览器（Chrome /Edge）高效使用 - 内部命令/快捷键/启动参数