Python 基础进阶（三）：从函数、类到 asyncio 与 FastAPI 后端开发实战

目录

一、函数：将重复逻辑封装起来

1.1 默认参数

1.2 关键字参数

1.3 可变参数 *args 与 **kwargs

1.4 小案例：订单金额计算

二、类与对象：让数据和行为放在一起

2.1 __init__ 是什么

2.2 类和对象的关系

2.3 小案例：简单任务类

三、装饰器：不修改原函数也能增强功能

3.1 最简单的装饰器

3.2 支持任意参数的装饰器

3.3 FastAPI 中的装饰器

四、迭代器与生成器

4.1 什么是迭代

4.2 什么是迭代器

4.3 自定义迭代器

4.4 什么是生成器

4.5 小案例：分批读取数据

五、asyncio：Python 的异步编程基础

5.1 同步执行的问题

5.2 异步执行

5.3 async 与 await

5.4 一个重要误区

六、FastAPI：从零写一个任务管理后端

6.1 安装 FastAPI

6.2 第一个 FastAPI 接口

6.3 路径参数

6.4 查询参数

七、FastAPI 小项目：任务管理 API

7.1 Pydantic 模型的作用

7.2 测试新增任务接口

7.3 测试修改任务状态

7.4 HTTPException 的作用

八、FastAPI 与 asyncio 的关系

总结

在前两篇中，我们学习了 Python 常用容器，例如列表、元组、字符串、集合和字典。 容器解决的是“如何保存一批数据”的问题；而接下来，我们需要学习“如何组织和处理这些数据”。 本篇将继续学习 Python 函数、类、装饰器、迭代器、生成器、asyncio，并重点入门 FastAPI 后端开发。

一、函数：将重复逻辑封装起来

函数可以理解为一段可重复使用的代码。例如，我们需要多次计算商品订单总价，如果每次都手动写计算逻辑，代码会重复且不利于维护。

price = 99.9
count = 2
total = price * count

print(total)

可以把它封装成函数：

def calculate_total(price, count):
return price * count

result = calculate_total(99.9, 2)

print(result)

输出：

199.8

函数由以下部分组成：

def 函数名(参数):
函数体
return 返回值

其中：

• def：定义函数的关键字；
• 函数名：建议使用小写字母和下划线；
• 参数：函数需要接收的数据；
• return：将结果返回给调用者。

1.1 默认参数

有时函数的某个参数通常使用固定值，可以给它设置默认值。

def greet(name, message="你好"):
return f"{message}，{name}！"

print(greet("张三"))
print(greet("李四", "早上好"))

输出：

你好，张三！
早上好，李四！

注意：带默认值的参数应该放在普通参数后面。

# 正确
def create_user(name, age=18):
pass
# 错误
def create_user(age=18, name):
pass

1.2 关键字参数

调用函数时，可以显式指定参数名称。

def create_user(name, age, city):
return {
"name": name,
"age": age,
"city": city
}

user = create_user(
city="西安",
name="小明",
age=20
)

print(user)

输出：

{'name': '小明', 'age': 20, 'city': '西安'}

关键字参数的好处是：参数顺序可以改变，代码可读性也更好。

1.3 可变参数 *args 与 **kwargs

当函数接收的参数个数不固定时，可以使用可变参数。*numbers 会把传入的多个位置参数收集为元组。

def calculate_sum(*numbers):
return sum(numbers)

print(calculate_sum(1, 2, 3))
print(calculate_sum(10, 20, 30, 40))

**kwargs 会把多个关键字参数收集为字典。

def print_user_info(**kwargs):
for key, value in kwargs.items():
print(f"{key}: {value}")

print_user_info(
name="小明",
age=20,
city="西安"
)

这和前面学习的容器知识正好对应：

*args -> 元组 tuple
**kwargs -> 字典 dict

1.4 小案例：订单金额计算

def calculate_order_total(price, count, discount=1.0):
"""计算订单最终金额。

price: 商品单价
count: 商品数量
discount: 折扣，默认不打折
"""
original_total = price * count
final_total = original_total * discount

return round(final_total, 2) #保留两位小数

total = calculate_order_total(
price=299.0,
count=2,
discount=0.85
)

print(f"订单最终金额：{total} 元")

输出：

订单最终金额：508.3 元

二、类与对象：让数据和行为放在一起

函数适合处理单一逻辑。但当一个事物既有数据、又有行为时，使用类会更合适。

例如，一个学生拥有：

姓名
年龄
分数

同时还可以：

自我介绍
修改分数
判断是否及格

这时就可以使用类。

class Student:
def __init__(self, name, age, score):
self.name = name
self.age = age
self.score = score

def introduce(self):
print(f"大家好，我叫 {self.name}，今年 {self.age} 岁。")

def is_passed(self):
return self.score >= 60

def update_score(self, new_score):
self.score = new_score

创建对象：

student = Student("小明", 20, 88)

student.introduce()

print(student.is_passed())

student.update_score(95)

print(student.score)

输出：

大家好，我叫 小明，今年 20 岁。
True
95

2.1 __init__ 是什么

__init__ 是初始化方法。当我们创建对象时：

student = Student("小明", 20, 88)

Python 会自动调用：

__init__(self, "小明", 20, 88)

其中：

self.name = name

表示将传入的 name 保存到当前对象中。self 可以理解为“当前创建出来的对象本身”。

2.2 类和对象的关系

Student 类
↓
创建对象
↓
student1、student2、student3

例如：

student1 = Student("小明", 20, 88)
student2 = Student("小红", 21, 56)

print(student1.name)
print(student2.name)

print(student1.is_passed())
print(student2.is_passed())

每个对象都有自己的属性数据。

2.3 小案例：简单任务类

后面使用 FastAPI 编写后端时，也会频繁处理“任务”“用户”“订单”等对象。

class Task:
def __init__(self, task_id, title, completed=False):
self.task_id = task_id
self.title = title
self.completed = completed

def finish(self):
self.completed = True

def to_dict(self):
return {
"id": self.task_id,
"title": self.title,
"completed": self.completed
}

task = Task(1, "学习 FastAPI")

print(task.to_dict())

task.finish()

print(task.to_dict())

输出：

{'id': 1, 'title': '学习 FastAPI', 'completed': False}
{'id': 1, 'title': '学习 FastAPI', 'completed': True}

三、装饰器：不修改原函数也能增强功能

装饰器是 Python 中一个非常重要的功能。它可以在不修改原函数代码的前提下，为函数增加额外能力。

常见使用场景：

• 记录日志；
• 统计函数执行时间；
• 登录校验；
• 权限校验；
• 请求参数校验；
• 缓存处理；
• Web 接口路由注册。

3.1 最简单的装饰器

def log_decorator(func):
def wrapper():
print("函数开始执行")
func()
print("函数执行结束")

return wrapper

@log_decorator
def say_hello():
print("你好，Python！")

say_hello()

输出：

函数开始执行
你好，Python！
函数执行结束

这一句：

@log_decorator

本质上等价于：

say_hello = log_decorator(say_hello)

3.2 支持任意参数的装饰器

实际函数往往有参数，因此装饰器通常写成下面这样：

from functools import wraps

def log_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"正在调用函数：{func.__name__}")

result = func(*args, **kwargs)

print(f"函数执行完成，结果：{result}")

return result

return wrapper

使用：

@log_decorator
def add(a, b):
return a + b

print(add(3, 5))

输出：

正在调用函数：add
函数执行完成，结果：8
8

@wraps(func) 的作用是保留原函数的名称、注释等元信息，是编写装饰器时的推荐写法。

3.3 FastAPI 中的装饰器

后面会看到，FastAPI 最常见的写法就是装饰器：

@app.get("/tasks")
async def get_tasks():
return []

其中：

@app.get("/tasks")

就是一个装饰器。它的作用是告诉 FastAPI：

当客户端发送 GET /tasks 请求时，
请执行 get_tasks 函数。

所以理解装饰器，对学习 FastAPI 非常重要。

四、迭代器与生成器

4.1 什么是迭代

遍历列表时，我们已经使用过 for 循环：

numbers = [10, 20, 30]

for number in numbers:
print(number)

这里的列表是可迭代对象。

常见可迭代对象包括：

• list；
• tuple；
• str；
• set；
• dict；
• 文件对象；
• 生成器。

4.2 什么是迭代器

迭代器是一个可以不断通过 next() 获取下一个元素的对象。

numbers = [10, 20, 30]

iterator = iter(numbers)

print(next(iterator))
print(next(iterator))
print(next(iterator))

输出：

10
20
30

当没有更多元素时，再调用 next() 会抛出异常：

StopIteration

for 循环本质上也是不断调用 next()，只是 Python 自动帮我们处理了停止异常。

4.3 自定义迭代器

class CountDown:
def __init__(self, start):
self.current = start

def __iter__(self):
return self

def __next__(self):
if self.current <= 0:
raise StopIteration

value = self.current
self.current -= 1

return value

使用：

for number in CountDown(3):
print(number)

输出：

3
2
1

4.4 什么是生成器

生成器是更简单的迭代器写法。它使用 yield 返回数据。

def count_down(start):
while start > 0:
yield start
start -= 1

调用：

generator = count_down(3)

print(next(generator))
print(next(generator))
print(next(generator))

或者直接使用 for：

for number in count_down(3):
print(number)

生成器最大的优势是节省内存。例如，如果需要处理一千万条数据：

numbers = [i for i in range(10_000_000)]

会一次性生成所有数据并占用内存。而生成器：

numbers = (i for i in range(10_000_000))

只有真正遍历到某个元素时，才生成那个元素。

4.5 小案例：分批读取数据

def batch_data(data, batch_size):
for index in range(0, len(data), batch_size):
yield data[index:index + batch_size]

users = [
"用户1", "用户2", "用户3",
"用户4", "用户5", "用户6",
"用户7"
]

for batch in batch_data(users, 3):
print(batch)

输出：

['用户1', '用户2', '用户3']
['用户4', '用户5', '用户6']
['用户7']

这种“分批处理”的思路，在爬虫、数据库批量写入、文件处理、消息队列消费等场景中很常见。

五、asyncio：Python 的异步编程基础

学习 FastAPI 前，必须先理解 asyncio 的基础。

5.1 同步执行的问题

假设有两个任务：

import time

def download_file():
print("开始下载文件")
time.sleep(3)
print("文件下载完成")

def send_message():
print("开始发送消息")
time.sleep(2)
print("消息发送完成")

download_file()
send_message()

总耗时约为：

3 秒 + 2 秒 = 5 秒

因为第一个任务没有结束，第二个任务不能开始。

5.2 异步执行

import asyncio

async def download_file():
print("开始下载文件")
await asyncio.sleep(3)
print("文件下载完成")

async def send_message():
print("开始发送消息")
await asyncio.sleep(2)
print("消息发送完成")

async def main():
await asyncio.gather(
download_file(),
send_message()
)

asyncio.run(main())

总耗时约为：

3 秒

因为两个任务在等待 I/O 时可以交替执行。

5.3 async 与 await

async def task():
await asyncio.sleep(1)

含义：

• async def：定义协程函数；
• await：等待一个异步操作完成，同时让出执行权；
• asyncio.run()：启动事件循环并执行协程；
• asyncio.gather()：并发等待多个协程完成。

5.4 一个重要误区

异步不是“让所有代码都变快”。如果函数是 CPU 密集型任务，例如：

def calculate():
total = 0

for i in range(100_000_000):
total += i

return total

即使写成 async def，也不会自动加速。

异步更适合 I/O 密集型场景，例如：

• 请求第三方接口；
• 查询数据库；
• 读写文件；
• 等待 Redis；
• 等待消息队列；
• 等待网络响应。

在异步函数中，不要使用阻塞式的：

time.sleep(3)

而应该使用：

await asyncio.sleep(3)

六、FastAPI：从零写一个任务管理后端

FastAPI 是一个现代 Python Web 框架，适合开发：

• RESTful API；
• 前后端分离项目；
• 管理后台接口；
• 微服务；
• AI 应用后端；
• 文件上传服务；
• 数据分析接口；
• 模型推理接口。

FastAPI 的几个优点：

• 基于 Python 类型注解；
• 自动校验请求参数；
• 自动生成 Swagger 文档；
• 原生支持异步；
• 开发体验友好；
• 适合构建高性能 API。

6.1 安装 FastAPI

建议先创建虚拟环境：

python -m venv .venv

Windows 激活虚拟环境：

.venv\\Scripts\\activate

安装 FastAPI：

python -m pip install "fastapi[standard]"

项目目录可以先保持简单：

python_fastapi_demo/
├── main.py
└── requirements.txt

requirements.txt 内容：

fastapi[standard]

6.2 第一个 FastAPI 接口

创建 main.py：

from fastapi import FastAPI

app = FastAPI(
title="任务管理 API",
description="用于学习 FastAPI 的简单后端项目",
version="1.0.0"
)

@app.get("/")
async def read_root():
return {
"message": "FastAPI 服务启动成功"
}

启动服务：

fastapi dev main.py

也可以使用传统的 Uvicorn 命令：

uvicorn main:app –reload

其中：

main -> main.py 文件
app -> FastAPI() 创建的对象
–reload -> 修改代码后自动重启服务

启动后访问：

http://127.0.0.1:8000/

返回：

{
"message": "FastAPI 服务启动成功"
}

FastAPI 还会自动生成接口文档：

http://127.0.0.1:8000/docs

这是 Swagger UI 页面，可以直接在浏览器中测试接口。FastAPI 的官方文档也提供了使用 Python 基础镜像构建容器化服务的标准目录与启动方式，后续可以把本篇的小项目直接容器化部署。FastAPI 官方部署文档

6.3 路径参数

路径参数是 URL 路径中的动态部分。

from fastapi import FastAPI

app = FastAPI()

@app.get("/tasks/{task_id}")
async def get_task(task_id: int):
return {
"task_id": task_id,
"message": "查询任务成功"
}

访问：

http://127.0.0.1:8000/tasks/1

返回：

{
"task_id": 1,
"message": "查询任务成功"
}

这里：

task_id: int

不仅是类型提示，也是请求参数校验。如果访问：

/tasks/abc

FastAPI 会自动返回参数校验错误，因为 abc 不能转换为整数。

6.4 查询参数

查询参数通常放在 URL 的 ? 后面。

@app.get("/tasks")
async def get_tasks(
keyword: str | None = None,
completed: bool | None = None
):
return {
"keyword": keyword,
"completed": completed
}

访问：

http://127.0.0.1:8000/tasks?keyword=FastAPI&completed=false

返回：

{
"keyword": "FastAPI",
"completed": false
}

路径参数和查询参数的区别：

七、FastAPI 小项目：任务管理 API

下面使用内存列表模拟数据库，实现一个简单任务管理系统。

支持功能：

• 查询全部任务；
• 根据 ID 查询任务；
• 新增任务；
• 修改任务状态；
• 删除任务。

完整 main.py：

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

app = FastAPI(
title="任务管理 API",
description="FastAPI 入门实战项目",
version="1.0.0"
)

tasks = [
{
"id": 1,
"title": "学习 Python 函数",
"completed": True
},
{
"id": 2,
"title": "学习 FastAPI",
"completed": False
}
]

class TaskCreate(BaseModel):
title: str = Field(
min_length=1,
max_length=100,
description="任务标题"
)

class TaskUpdate(BaseModel):
completed: bool = Field(
description="任务是否完成"
)

@app.get("/")
async def read_root():
return {
"message": "任务管理 API 服务运行中"
}

@app.get("/tasks")
async def get_tasks(completed: bool | None = None):
"""查询任务列表。"""

if completed is None:
return tasks

return [
task for task in tasks
if task["completed"] == completed
]

@app.get("/tasks/{task_id}")
async def get_task(task_id: int):
"""根据任务 ID 查询任务。"""

for task in tasks:
if task["id"] == task_id:
return task

raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="任务不存在"
)

@app.post(
"/tasks",
status_code=status.HTTP_201_CREATED
)
async def create_task(task: TaskCreate):
"""新增任务。"""

new_task = {
"id": len(tasks) + 1,
"title": task.title,
"completed": False
}

tasks.append(new_task)

return new_task

@app.patch("/tasks/{task_id}")
async def update_task(
task_id: int,
task_data: TaskUpdate
):
"""修改任务完成状态。"""

for task in tasks:
if task["id"] == task_id:
task["completed"] = task_data.completed
return task

raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="任务不存在"
)

@app.delete("/tasks/{task_id}")
async def delete_task(task_id: int):
"""删除任务。"""

for index, task in enumerate(tasks):
if task["id"] == task_id:
deleted_task = tasks.pop(index)

return {
"message": "任务删除成功",
"task": deleted_task
}

raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="任务不存在"
)

7.1 Pydantic 模型的作用

class TaskCreate(BaseModel):
title: str = Field(min_length=1, max_length=100)

这个类用于接收新增任务时的 JSON 数据。当客户端发送：

{
"title": "学习 asyncio"
}

FastAPI 会自动：

如果客户端发送：

{
"title": ""
}

FastAPI 会自动返回参数校验错误，不需要手动写大量 if 判断。

7.2 测试新增任务接口

请求地址：

POST http://127.0.0.1:8000/tasks

请求体：

{
"title": "学习 FastAPI 请求体"
}

返回：

{
"id": 3,
"title": "学习 FastAPI 请求体",
"completed": false
}

7.3 测试修改任务状态

请求地址：

PATCH http://127.0.0.1:8000/tasks/2

请求体：

{
"completed": true
}

返回：

{
"id": 2,
"title": "学习 FastAPI",
"completed": true
}

7.4 HTTPException 的作用

当任务不存在时：

raise HTTPException(
status_code=404,
detail="任务不存在"
)

客户端会收到：

{
"detail": "任务不存在"
}

常见状态码：

八、FastAPI 与 asyncio 的关系

你会发现，FastAPI 接口经常写成：

@app.get("/tasks")
async def get_tasks():
return tasks

这里的 async def 表示该接口可以使用异步方式执行。例如，模拟一个需要等待数据库查询的接口：

import asyncio

@app.get("/slow-tasks")
async def get_slow_tasks():
await asyncio.sleep(2)

return {
"message": "模拟数据库查询完成",
"data": tasks
}

这里使用：

await asyncio.sleep(2)

表示当前请求等待时，不会一直阻塞整个服务。FastAPI 可以在这段等待时间里处理其他请求。

但要注意：

import time

@app.get("/bad-example")
async def bad_example():
time.sleep(2)

return {"message": "不推荐这样写"}

虽然接口定义成了 async def，但 time.sleep() 仍然是阻塞操作，会影响异步性能。

正确思路：

因此可以记住一句话：

async def 本身不会让代码变快；真正决定是否异步的是内部调用的操作是否支持 await。

总结

本篇从 Python 容器继续向后学习了 Python 后端开发的重要基础。

• 函数：封装重复逻辑；
• 类与对象：组织数据和行为；
• 装饰器：为函数增加额外能力；
• 迭代器：按顺序获取数据；
• 生成器：节省内存地生成大量数据；
• asyncio：处理 I/O 密集型异步任务；
• FastAPI：快速构建现代 Web API；
• Pydantic：自动校验请求参数；
• HTTPException：规范返回接口异常；
• Swagger：通过 /docs 自动测试接口。

本篇的任务管理 API 目前使用的是内存列表保存数据，服务重启后数据会丢失。

下一篇可以继续将任务管理 API 升级为：

FastAPI
+ SQLAlchemy
+ MySQL / SQLite
+ 用户注册登录
+ JWT 权限认证
+ Docker 容器化部署