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

(Python)[特殊字符] 基于Flask/FastAPI的RESTful API服务 + 数据库 + 缓存 + 简单前端 (Python项目)

目录

前言:

源代码:

__init__.py

database.py

models.py

schemas.py

routers.py

.env

.gitignore

main.py

代码解析:

📦 第一步:创建项目结构

🔧 第二步:安装依赖

⚙️ 第三步:环境变量配置 (.env)

🛡️ 第四步:创建 .gitignore

这是什么?

工作原理

🧩 第五步:配置读取 (app/init.py)

这是什么?

为什么需要它?(核心作用)

这行代码 load_dotenv() 做了什么?

为什么放在这里?

⚙️ 第六步:数据库配置 (app/database.py)

🔧 关键概念详解

1. 数据库引擎 (Engine)

2. 会话 (Session)

3. ORM (对象关系映射)

4. 连接池 (Connection Pool)

📊 第七步:定义数据模型 (app/models.py)

📊 数据模型 ↔ 数据库表 映射关系

🔍 核心概念详解

1. Column 类型详解

2. 列参数详解

3. 继承 Base 的重要性

🧪 实际数据库表结构

📝 第八步:数据验证模型 (app/schemas.py)

📊 各模型用途对比

🔍 核心概念详解

1. Pydantic 是什么?

2. 为什么需要多个模型?

3. 类型注解的重要性

🧪 实际工作流程

创建 Todo 的完整流程

验证错误示例

🚦 第九步:API路由 (app/routers.py)

📊 HTTP 方法与 CRUD 操作对应关系

🔍 核心概念详解

1. APIRouter 是什么?

2. 依赖注入 (Dependency Injection)

3. 路径参数 vs 查询参数

🧪 实际请求示例

创建 Todo (POST)

获取 Todo (GET)

获取分页列表 (GET with query params)

更新 Todo (PUT)

删除 Todo (DELETE)

🛠️ 数据库操作详解

查询方法对比

错误处理

💡 最佳实践建议

1. 路由组织

2. 错误处理

🚀 第十步:主应用入口 (main.py)

📊 FastAPI 应用启动流程

🔍 核心概念详解

1. FastAPI 应用实例 (app = FastAPI())

2. 路由注册 (app.include_router())

3. 启动事件 (@app.on_event("startup"))

4. Uvicorn 服务器 (uvicorn.run())

方式:直接运行 Python 文件

🌐 网络配置详解

host 参数说明

port 参数说明

🧪 测试你的应用

1. 启动应用

2. 测试健康检查

3. 测试API文档

4. 测试数据库连接

🛠️ 第十一步:创建MySQL数据库

🚦 第十二步:运行应用

🧪 测试API

创建Todo

获取Todo

获取所有Todo

更新Todo

删除Todo


前言:

运行之后访问:http://localhost:8000/docs,进行操作

源代码:

上图是文件位置

__init__.py

# app\__init__.py
# 初始化
from dotenv import load_dotenv
# 加载。env文件中的环境变量
load_dotenv()

database.py

# app\database.by
# 数据库配置文件# 0.导入必要的库
from sqlalchemy import text
from sqlalchemy import create_engine    # 创建数据库引擎的核心类
from sqlalchemy.ext.declarative import declarative_base # 用于创建数据模型基类
from sqlalchemy.orm import sessionmaker # 用于创建数据库会话
import os   # 用于访问操作系统环境变量# 1.获取数据库配置
# 从环境变量中读取数据库主机地址(通常是localhost或服务器IP)
DB_HOST=os.getenv("DB_HOST")    # os.getenv()是获取环境变量的值
# 从环境变量中读取数据库端口号(MySQL通常是3306)
DB_PORT=os.getenv("DB_PORT")
# 从环境变量中读取数据库用户名
DB_USER=os.getenv("DB_USER")
# 从环境变量中读取数据库密码
DB_PASSWORD=os.getenv("DB_PASSWORD")
# 从环境变量中读取数据库名
DB_NAME=os.getenv("DB_NAME")# 2.构建数据库连接字符串
# 构建数据库URL
# 格式: mysql+pymysql://用户名:密码@主机地址:端口/数据库名
SQLALCHEMY_DATABASE_URL=f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
# 解释:
# 1. `mysql+pymysql`:表示使用MySQL数据库,并通过pymysql驱动连接
# 2. `{DB_USER}:{DB_PASSWORD}`:用户名和密码
# 3. `@{DB_HOST}:{DB_PORT}`:数据库服务器地址和端口
# 4. `/{DB_NAME}`:要连接的数据库名称# 3.创建数据库引擎(这是与数据库通讯的核心)
engine=create_engine(SQLALCHEMY_DATABASE_URL,    # 使用上面的链接# 连接池设置(提高性能)pool_size=5,    # 保持5个常驻数据库连接max_overflow=10,    # 当连接不足时,最多可创建10个临时连接pool_recycle=3600,  # 每隔3600秒(1小时)回收并重建连接
)
# 为什么要连接池?
# 每次请求都新建数据库连接很慢,连接池保留一些"预热"的连接,
# 可以快速响应请求,提高性能# 4.创建会话工厂
SessionLocal=sessionmaker(autocommit=False,   # 关闭自动提交(推荐手动控制事务)autoflush=False,    # 关闭自动刷新(提高性能)bind=engine,    # 绑定到上面创建的引擎
)
# 会话(Session)是什么?
# 会话是数据库操作的"工作区",所有数据库操作(增删改查)都通过会话进行
# 每个API请求都会创建一个新的会话,请求结束后自动关闭# 5.声明基类
# 创建所有数据模型继承的基类
Base=declarative_base()
# 这个Base类有什么用?
# 1. 所有数据模型类都会继承这个Base
# 2. 它知道如何将Python类映射到数据库表
# 3. 它提供了创建表的方法(在main.py中使用)# 6.创建依赖函数
"""生成一个数据库会话,请求结束后自动关闭这个函数会在每个API请求中被调用,为请求提供一个独立的数据库会话
"""
def get_db():# 从会话工厂创建一个新的会话db=SessionLocal()try:# 将会话提供给请求使用yield dbfinally:# 无论请求成功还是失败,最后都会关闭会话db.close()
# 为什么需要这个函数?
# FastAPI使用"依赖注入"系统,这个函数告诉FastAPI:
# "当API需要数据库会话时,调用我来获取一个"# 修改 test_connection 函数
def test_connection():"""测试数据库连接是否正常"""try:# 使用引擎直接连接with engine.connect() as conn:# 使用 text() 包装 SQL 语句result = conn.execute(text("SELECT 1"))print("✅ 数据库连接成功!")print("测试查询结果:", result.scalar())  # 应该返回1return Trueexcept Exception as e:print("❌ 数据库连接失败!")print("错误详情:", str(e))return False

models.py

# app\models.py
# 定义数据模型
# 导入必要的模块
from sqlalchemy import Column,Integer,Boolean,String
from .database import Base"""Todo数据模型类这个类对应数据库中的todos表每个Todo实例代表表中的一行数据
"""class Todo(Base):# 1.指定数据库表名__tablename__="todos"# __tablename__ 是特殊变量,告诉SQLAlchemy这个类对应哪个表# 表名通常使用复数形式(todos而不是todo)# 2. 定义id字段(主键)id=Column(Integer,primary_key=True,index=True)# Column: 定义一个数据库列# Integer: 整数类型# primary_key=True: 这是主键,唯一标识每行数据# index=True: 为此列创建索引,加快查询速度# 3. 定义title字段title=Column(String(100),nullable=False)# String(100): 最大长度100的字符串# nullable=False: 此字段不能为空(必须提供值)# 4. 定义description字段description=Column(String(500),nullable=True)# String(500): 最大长度500的字符串# nullable=True: 此字段可以为空(可选)# 5. 定义completed字段completed=Column(Boolean,default=False)# Boolean: 布尔类型(True/False)# default=False: 默认值为False

schemas.py

# app\schemas.py
# 数据验证模型
# 导入必要的模块
from pydantic import BaseModel  #pydantic的基础模型类
from typing import Optional #用于定义可选字段# 1.基础模型(TodoBase)(共享字段)
"""Todo基础模型包含所有Todo共有的字段定义其他模型可以继承这个基础模型
"""
class TodoBase(BaseModel):title:str   # 必须字段,字符串类型description:Optional[str]=None  # 可选字段,默认为None# 注意:这里没有id和completed字段,因为它们在创建时可能不需要# 2.创建Todo时使用的模型(TodoCreate)
"""创建Todo时使用的请求模型继承TodoBase的所有字段,并添加新字段
"""
class TodoCreate(TodoBase):completed: bool = False  # 可选字段,默认值为False# 这个模型用于 POST /todos/ 请求# 客户端可以发送:{"title": "学习", "description": "描述", "completed": true}# 3.更新Todo时使用的模型(TodoUpdata)
"""更新Todo时使用的请求模型允许部分更新(只更新提供的字段)
"""
class TodoUpdata(TodoBase):title:Optional[str]=Nonedescription:Optional[str]=Nonecompleted:Optional[bool]=None# 这个模型用于 PUT /todos/{id} 请求# 客户端可以只发送要更新的字段:{"completed": true}# 4.响应模型Todo(返回给客户的数据)
"""返回给客户端的Todo响应模型包含所有字段,包括数据库生成的id
"""
class Todo(TodoBase):id:int  # 数据库自动生成的IDcompleted:bool  # 完成状态# Pydantic配置 - 允许从ORM对象转换class Config:from_attributes=True    # 旧版本叫 orm_mode = True# 这个配置的作用:# 允许 FastAPI 自动将数据库模型实例(SQLAlchemy)转换为Pydantic模型# 例如:db_todo(SQLAlchemy对象) -> Todo(Pydantic对象)

routers.py

# app/routers.py
# API路由(FastAPI核心)# 导入必要的模块
from fastapi import APIRouter, HTTPException, status, Depends
from sqlalchemy.orm import Session  # 数据库会话类型
from typing import List    # 用于类型注释(List)
from . import models, schemas   # 导入自定义模块
from .database import get_db  # 正确导入方式# 1. 创建路由实例
router = APIRouter(prefix="/todos",    # 所有路由都会自动添加这个前缀tags=["todos"]   # 在API文档中分组显示
)# 2. 创建Todo - POST /todos/
@router.post("/",response_model=schemas.Todo,   # 指定响应模型status_code=status.HTTP_201_CREATED)   # 指定HTTP状态码
def create_todo(todo: schemas.TodoCreate,    # 请求体(自动验证)db: Session = Depends(get_db)   
):"""创建新的Todo项- **title**: Todo标题 (必需)- **description**: 详细描述 (可选)- **completed**: 是否完成 (默认False)"""# 将Pydantic模型转换为SQLAlchemy模型db_todo = models.Todo(title=todo.title,description=todo.description,completed=todo.completed,)# 添加到数据库会话db.add(db_todo)# 提交到数据库(永久保存)db.commit()# 刷新对象以获取数据库生成的ID等数据db.refresh(db_todo)return db_todo  # 自动转换为Todo响应模型# 3. 获取单个Todo - GET /todos/{todo_id}
@router.get("/{todo_id}", response_model=schemas.Todo)
def read_todo(todo_id: int,db: Session = Depends(get_db)  
):"""根据ID获取Todo项"""# 查询数据库:查找指定ID的Todo# .first() 返回第一个结果或Nonetodo = db.query(models.Todo).filter(models.Todo.id == todo_id).first()# 如果找不到id,返回404错误if todo is None:raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail="Todo not found"  # 错误详情)return todo# 4. 获取所有Todo - GET /todos/
@router.get("/", response_model=List[schemas.Todo])
def read_todos(skip: int = 0,  # 查询参数:跳过多少条(分页)limit: int = 100,  # 查询参数:返回多少条(分页)db: Session = Depends(get_db)  
):"""获取Todo列表- **skip**: 跳过前N条结果 (分页用)- **limit**: 返回结果数量限制 (分页用)"""# 查询所有Todo,使用分页参数todos = db.query(models.Todo).offset(skip).limit(limit).all()return todos# 5. 更新Todo - PUT /todos/{todo_id}
@router.put("/{todo_id}", response_model=schemas.Todo)
def update_todo(  todo_id: int,todo_update: schemas.TodoUpdata,  db: Session = Depends(get_db)  
):"""更新Todo项只需提供需要更新的字段"""# 先获取现有的Tododb_todo = db.query(models.Todo).filter(models.Todo.id == todo_id).first()  if db_todo is None:  # 检查是否找到idraise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail="Todo not found")# 获取客户端提供的更新数据(只包含实际发送的字段)update_data = todo_update.model_dump(exclude_unset=True)  # 遍历更新数据,设置到数据库对象中for key, value in update_data.items():setattr(db_todo, key, value)  # 动态设置属性# 提交更改到数据库db.commit()# 刷新对象(获取最新状态)db.refresh(db_todo)return db_todo# 6. 删除Todo - DELETE /todos/{todo_id}
@router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_todo(todo_id: int,db: Session = Depends(get_db)  
):"""删除Todo项返回204状态码(成功且无内容返回)"""# 查找要删除的Todotodo = db.query(models.Todo).filter(models.Todo.id == todo_id).first()if todo is None:raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail="Todo not found")# 从数据库删除db.delete(todo)db.commit()# 204状态码不需要返回内容return

.env

#环境变量配置
#数据库链接配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=123456(你自己的密码)
DB_NAME=todo_app

.gitignore

#忽略环境变量文件
.env
#忽略Python缓存
__pycache__/
*.pyc
#忽略IDE配置文件
.idea/
.vscode/
#忽略虚拟环境
venv/

main.py

# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager
from app.database import test_connection, Base, engine
from app import routers# 新的生命周期事件处理方式
@asynccontextmanager
async def lifespan(app: FastAPI):# 启动逻辑if test_connection():Base.metadata.create_all(bind=engine)print("✅ 数据库表已初始化")else:print("❌ 无法初始化数据库")yield# 关闭逻辑(可选)# 例如:关闭数据库连接池等# 创建 FastAPI 应用实例,使用新的生命周期处理器
app = FastAPI(title="Todo API",description="一个简单的Todo应用",version="1.0.0",lifespan=lifespan  # 使用新的生命周期处理器
)# 包含路由
app.include_router(routers.router)# 根路由
@app.get("/")
def read_root():return {"message": "Todo API 正在运行"}if __name__ == "__main__":import uvicornuvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

代码解析:

📦 第一步:创建项目结构

在 PyCharm 中创建以下文件结构:

my_fastapi_project/
├── .env                # 环境变量配置文件
├── .gitignore          # Git忽略文件
├── main.py             # 应用入口
├── requirements.txt    # 依赖列表
└── app/                # 应用核心代码├── __init__.py├── database.py     # 数据库连接├── models.py       # 数据模型├── schemas.py      # 数据验证└── routers.py      # API路由

🔧 第二步:安装依赖

在终端执行:

pip install fastapi uvicorn sqlalchemy pymysql python-dotenv pydantic-settings

⚙️ 第三步:环境变量配置 (.env)

# 数据库连接配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=your_password
DB_NAME=todo_app

说明

  1. 这个文件存储敏感信息(密码等)

  2. 文件名 .env 开头的点表示隐藏文件

  3. 每行格式:变量名=值

  4. 不要提交到 Git(加到 .gitignore

🛡️ 第四步:创建 .gitignore

这是什么?

.gitignore 是一个纯文本文件,用于告诉 Git 版本控制系统哪些文件或目录不应该被跟踪和上传到代码仓库。

为什么需要它?(核心作用)

作用示例不使用的后果
保护敏感信息防止数据库密码泄露黑客获取你的数据库密码
避免垃圾文件忽略Python缓存文件仓库被无用文件污染
保持仓库整洁忽略IDE配置文件团队协作时配置冲突
节省存储空间忽略虚拟环境文件仓库体积巨大,下载缓慢

工作原理

当你执行 git add . 时:

  1. Git 会读取 .gitignore 文件

  2. 自动跳过列出的文件和目录

  3. 只添加真正需要版本控制的代码文件

🧩 第五步:配置读取 (app/init.py)

这是什么?

__init__.py 是一个特殊的Python文件,有两个主要作用:

  1. 标识 app 目录是一个 Python 包(package)

  2. 初始化包的配置和环境

为什么需要它?(核心作用)

作用说明必要性
包标识告诉Python这个目录是一个可导入的包必须存在(可以是空文件)
初始化代码放置需要在导入包时运行的代码可选但推荐
统一入口集中管理包级别的配置提高可维护性

这行代码 load_dotenv() 做了什么?

  1. 查找文件:自动在项目根目录寻找 .env 文件

  2. 读取内容:解析 KEY=VALUE 格式的配置

  3. 注入环境:将这些配置添加到Python的 os.environ 中

  4. 全局可用:后续代码可以通过 os.getenv("KEY") 获取值

为什么放在这里?

  1. 最先执行:当任何 app 模块被导入时,__init__.py 会最先运行

  2. 确保可用:保证环境变量在其他代码执行前已加载

  3. 单点配置:所有环境配置集中在一个地方管理

第四步和第五步工作流程

⚙️ 第六步:数据库配置 (app/database.py)

🔧 关键概念详解

1. 数据库引擎 (Engine)

  • 作用:管理数据库连接的核心组件

  • 特点

    • 创建一次,整个应用共享

    • 管理连接池

    • 将SQL语句转换为数据库能理解的格式

2. 会话 (Session)

  • 作用:数据库操作的"工作区"

  • 生命周期

    • 请求开始时创建

    • 请求结束时关闭

  • 重要特性

    • 每个请求独立会话(避免数据混乱)

    • 跟踪对象状态变化(用于自动更新)

    • 管理事务(要么全部成功,要么全部失败)

3. ORM (对象关系映射)

  • 原理:将数据库表映射为Python类

  • 示例

    # 数据库中的todos表
    class Todo(Base):  # ← 继承自Base__tablename__ = 'todos'id = Column(Integer, primary_key=True)title = Column(String(100))
  • 优势

    • 用Python对象操作数据库,无需写SQL

    • 自动处理数据类型转换

    • 提供高级查询接口

4. 连接池 (Connection Pool)

  • 解决的问题:频繁创建/销毁数据库连接性能差

  • 工作原理

    1. 启动时创建多个连接(pool_size)

    2. 请求来时从池中取出空闲连接

    3. 使用后归还到池中

    4. 定期检查连接有效性(pool_recycle)

📊 第七步:定义数据模型 (app/models.py)

📊 数据模型 ↔ 数据库表 映射关系

你的 Python 类会转换成这样的数据库表:

🔍 核心概念详解

1. Column 类型详解
数据类型说明对应MySQL类型
Integer整数INT
String(length)字符串(限制长度)VARCHAR(length)
Boolean布尔值BOOLEAN 或 TINYINT(1)
Text长文本TEXT
DateTime日期时间DATETIME
2. 列参数详解
参数作用示例
primary_key=True设置为主键id = Column(Integer, primary_key=True)
nullable=False不允许为空title = Column(String, nullable=False)
default=值设置默认值completed = Column(Boolean, default=False)
index=True创建索引id = Column(Integer, index=True)
unique=True值必须唯一email = Column(String, unique=True)
3. 继承 Base 的重要性
class Todo(Base):  # ← 必须继承Base
  • Base 是 declarative_base() 创建的基类

  • 它让 SQLAlchemy 知道这个类是一个数据模型

  • 它提供了将类映射到数据库表的能力

🧪 实际数据库表结构

你的 Todo 类会在 MySQL 中创建这样的表:

CREATE TABLE todos (id INT AUTO_INCREMENT PRIMARY KEY,title VARCHAR(100) NOT NULL,description VARCHAR(500) NULL,completed BOOLEAN DEFAULT FALSE
);

📝 第八步:数据验证模型 (app/schemas.py)

📊 各模型用途对比

模型用途示例JSON特点
TodoCreate创建新Todo{"title": "学习", "completed": false}包含创建所需字段
TodoUpdate更新现有Todo{"completed": true}所有字段可选
Todo响应数据{"id": 1, "title": "...", completed: false}包含所有字段

🔍 核心概念详解

1. Pydantic 是什么?

Pydantic 是一个数据验证库,它:

  • ✅ 自动验证输入数据的类型和格式

  • ✅ 提供清晰的错误信息

  • ✅ 支持默认值和可选字段

  • ✅ 自动生成API文档

2. 为什么需要多个模型?

3. 类型注解的重要性
# 这些不是普通注释,而是类型注解!
title: str                    # 必须是字符串
description: Optional[str]    # 可以是字符串或None
completed: bool = False       # 必须是布尔值,默认False

🧪 实际工作流程

创建 Todo 的完整流程
# 1. 客户端发送JSON
json_data = {"title": "学习Python","description": "完成FastAPI项目","completed": True
}# 2. FastAPI 自动验证(使用 TodoCreate 模型)
# - 检查title是否是字符串 ✅
# - 检查description是否是字符串或null ✅  
# - 检查completed是否是布尔值 ✅# 3. 如果验证失败,返回400错误
# 4. 如果验证成功,继续处理
验证错误示例

如果客户端发送:

{"title": 123,  # 数字而不是字符串"completed": "yes"  # 字符串而不是布尔值
}

FastAPI 会自动返回:

{"detail": [{"loc": ["body", "title"],"msg": "str type expected","type": "type_error.str"},{"loc": ["body", "completed"], "msg": "value could not be parsed to a boolean","type": "type_error.bool"}]
}

🚦 第九步:API路由 (app/routers.py)

📊 HTTP 方法与 CRUD 操作对应关系

HTTP 方法路由CRUD 操作说明
POST/todos/Create创建新资源
GET/todos/{id}Read读取单个资源
GET/todos/Read读取多个资源
PUT/todos/{id}Update更新资源
DELETE/todos/{id}Delete删除资源

🔍 核心概念详解

1. APIRouter 是什么?
router = APIRouter(prefix="/todos", tags=["todos"])
  • 作用:组织相关的路由端点

  • prefix:自动为所有路由添加前缀(避免重复写)

  • tags:在API文档中分组显示相关路由

2. 依赖注入 (Dependency Injection)
db: Session = Depends(database.get_db)
  • 作用:自动为每个请求提供数据库会话

  • 好处:避免手动管理数据库连接

  • 生命周期:请求开始时创建,请求结束时自动关闭

3. 路径参数 vs 查询参数
# 路径参数(在URL路径中)
@router.get("/{todo_id}")  # todo_id 是路径参数# 查询参数(在URL问号后)
def read_todos(skip: int = 0, limit: int = 100)  # skip和limit是查询参数

🧪 实际请求示例

创建 Todo (POST)
POST http://localhost:8000/todos/
Content-Type: application/json{"title": "学习FastAPI","description": "完成路由学习","completed": false
}
获取 Todo (GET)
GET http://localhost:8000/todos/1
获取分页列表 (GET with query params)
GET http://localhost:8000/todos/?skip=10&limit=5
更新 Todo (PUT)
PUT http://localhost:8000/todos/1
Content-Type: application/json{"completed": true
}
删除 Todo (DELETE)
DELETE http://localhost:8000/todos/1

🛠️ 数据库操作详解

查询方法对比
方法作用返回值
.all()获取所有结果列表
.first()获取第一个结果单个对象或None
.filter()添加过滤条件查询对象
.offset()跳过N条结果查询对象
.limit()限制返回数量查询对象
错误处理
# 手动抛出HTTP异常
raise HTTPException(status_code=404,detail="Todo not found"
)# 使用预定义状态码
from fastapi import status
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail="Todo not found"
)

💡 最佳实践建议

1. 路由组织
# 好的:清晰的路径结构
@router.get("/")          # 获取列表
@router.post("/")         # 创建项目
@router.get("/{item_id}") # 获取单个
@router.put("/{item_id}") # 更新
@router.delete("/{item_id}") # 删除
2. 错误处理
# 自定义错误处理
def get_todo_or_404(todo_id: int, db: Session):todo = db.query(models.Todo).filter(models.Todo.id == todo_id).first()if todo is None:raise HTTPException(status_code=404,detail=f"Todo with ID {todo_id} not found")return todo# 在路由中使用
@router.get("/{todo_id}")
def read_todo(todo_id: int, db: Session = Depends(get_db)):todo = get_todo_or_404(todo_id, db)return todo

🚀 第十步:主应用入口 (main.py)

📊 FastAPI 应用启动流程

🔍 核心概念详解

1. FastAPI 应用实例 (app = FastAPI())
app = FastAPI(title="Todo API",description="一个简单的Todo应用", version="1.0.0"
)
  • 作用:创建 FastAPI 应用的中央控制器

  • 参数

    • title:API 名称(显示在文档顶部)

    • description:API 描述(显示在文档中)

    • version:API 版本(用于区分不同版本)

2. 路由注册 (app.include_router())
app.include_router(routers.router)
  • 作用:将路由模块中的路由添加到主应用

  • 好处:模块化组织代码,避免所有路由堆在一个文件中

3. 启动事件 (@app.on_event("startup"))
@app.on_event("startup")
async def startup_event():# 初始化代码
  • 时机:在应用启动后,接收请求前执行

  • 用途:数据库初始化、缓存预热、配置加载等

  • 异步支持:可以使用 async def 处理异步操作

4. Uvicorn 服务器 (uvicorn.run())
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
  • 作用:启动 ASGI 服务器来运行 FastAPI 应用

  • 参数

    • main:app:应用位置(main.py 中的 app 实例)

    • host="0.0.0.0":监听所有网络接口(允许外部访问)

    • port=8000:使用 8000 端口

    • reload=True:开发模式,代码修改自动重启

方式:直接运行 Python 文件
python main.py

工作原理

python

if __name__ == "__main__":  # 当直接运行这个文件时import uvicornuvicorn.run(...)        # 启动服务器

🌐 网络配置详解

host 参数说明
host 值可访问性适用环境
"127.0.0.1"只能本机访问开发测试
"0.0.0.0"所有网络接口都可访问生产环境
"localhost"本地访问开发环境
port 参数说明
端口号常用用途备注
8000FastAPI 默认端口推荐使用
8080备用HTTP端口常见选择
80HTTP标准端口需要管理员权限
443HTTPS标准端口需要SSL证书

🧪 测试你的应用

1. 启动应用
python main.py

应该看到输出:

text

INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [1234]
INFO:     Started server process [5678]
✅ 数据库表已初始化
2. 测试健康检查

打开浏览器访问:http://localhost:8000/
应该看到:

{"message": "Todo API 正在运行"}
3. 测试API文档

访问:http://localhost:8000/docs
应该看到完整的交互式API文档

4. 测试数据库连接

如果数据库连接失败,你会看到:

❌ 无法初始化数据库

🛠️ 第十一步:创建MySQL数据库

如果你有Navicate,可以可视化创建数据库和登入

  1. 登录MySQL:

    mysql -u root -p
  2. 创建数据库:

    CREATE DATABASE todo_app CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
  3. 创建用户并授权(可选):

    CREATE USER 'todo_user'@'localhost' IDENTIFIED BY 'secure_password';
    GRANT ALL PRIVILEGES ON todo_app.* TO 'todo_user'@'localhost';
    FLUSH PRIVILEGES;

🚦 第十二步:运行应用

在终端执行:

python main.py

访问:

  • API文档:http://localhost:8000/docs

  • 健康检查:http://localhost:8000/

🧪 测试API

创建Todo

POST http://localhost:8000/todos/
Content-Type: application/json{"title": "学习FastAPI","description": "完成第一个API项目"
}

获取Todo

GET http://localhost:8000/todos/1

获取所有Todo

GET http://localhost:8000/todos/

更新Todo

PUT http://localhost:8000/todos/1
Content-Type: application/json{"completed": true
}

删除Todo

DELETE http://localhost:8000/todos/1

注:该代码是本人自己所写,可能不够好,不够简便,欢迎大家指出我的不足之处。如果遇见看不懂的地方,可以在评论区打出来,进行讨论,或者联系我。上述内容全是我自己理解的,如果你有别的想法,或者认为我的理解不对,欢迎指出!!!如果可以,可以点一个免费的赞支持一下吗?谢谢各位彦祖亦菲!!!!

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

相关文章:

  • 【报错】Please do not run this script with sudo bash
  • 自建开发工具IDE(一)之拖找排版—仙盟创梦IDE
  • 网络编程5(HTTPS)
  • CentOS7.9中安装Harbor以及配置https
  • STM32 定时器(级联实现32位定时器)
  • 机器学习数据预处理全流程:从缺失值处理到特征编码
  • Python 全栈开发常用命令
  • 安路EF2系列芯片单口ram ip核使用方法
  • 阿里通义千问Qwen-Long 快速文档解析
  • 【Git】执行命令时要求输入Username、Password
  • RabbitMQ:SpringAMQP Topic Exchange(主题交换机)
  • Flink双流join
  • 【黑客技术零基础入门】PHP环境搭建、安装Apache、安装与配置MySQL(非常详细)零基础入门到精通,收藏这一篇就够
  • (认识异常)
  • 建模工具Sparx EA的多视图协作教程
  • [系统架构设计师]面向服务架构设计理论与实践(十五)
  • Shader学习路线
  • C++ MFC/BCG编程:文件对话框(CFileDialog、CFolderPickerDialog)
  • 【免费AI文档助手开发实战系列】基于正则表达式的PDF脱敏python服务构建(一)
  • 国产化PDF处理控件Spire.PDF教程:如何使用 Python 添加水印到 PDF
  • 太阳光模拟器在无人机老化测试中的应用
  • JVM参数优化
  • Nacos-8--分析一下nacos中的AP和CP模式
  • InfoNES模拟器HarmonyOS移植指南
  • SpringAI接入openAI配置出现的问题全解析
  • hadoop技术栈(九)Hbase替代方案
  • 深入理解计算机系统
  • 9-302 家里网能搜出两个ip, 无法联大堂监控室
  • LangChain —多模态 / 多源上下文管理
  • 银河麒麟V10一键安装Oracle 11g脚本分享