初识python中的FastAPI
·
一、FastAPI简介与安装
1.1 什么是FastAPI？
FastAPI是一个基于Python 3.8+的现代化Web框架，专门用于构建高性能的RESTful API。它基于Starlette（异步框架）和Pydantic（数据验证），利用Python类型提示提供自动API文档生成和强大的输入验证。
1.2 核心特点
-
高性能：支持异步编程，处理并发请求效率高
-
自动文档：自动生成Swagger UI和ReDoc交互式文档
-
类型安全：利用Python类型提示进行数据验证
-
开发友好：代码简洁，学习曲线平缓
1.3 安装FastAPI
# 安装FastAPI框架
pip install fastapi
# 安装ASGI服务器（用于运行FastAPI应用）
pip install "uvicorn[standard]"二、FastAPI基础概念
2.1 HTTP请求基础
一个完整的HTTP请求URL包含以下部分：
http://127.0.0.1:8000/hello/张三?username=bobo&userpass=bobo123
├── 协议部分: http/https
├── 域名部分: IP地址+端口号
├── 页面部分: 请求路径
└── 参数部分: 查询参数GET vs POST请求区别：
-
GET：参数暴露在URL中，速度快，安全性较低
-
POST：参数隐藏，速度较慢，安全性较高
2.2 序列化与反序列化
import json
# 示例数据
data = [
{"id": "s1001", "name": "张三"},
{"id": "s1002", "name": "李四"},
{"id": "s1003", "name": "王五"}
]
# 序列化：JSON → 字符串
data_str = json.dumps(data)
print(f"序列化结果: {data_str}, 类型: {type(data_str)}")
# 反序列化：字符串 → JSON
data_json = json.loads(data_str)
print(f"反序列化结果: {data_json}, 类型: {type(data_json)}")三、FastAPI基础使用
3.1 创建第一个FastAPI应用
from fastapi import FastAPI
import uvicorn
from typing import Union
# 创建FastAPI实例
app = FastAPI()
# 模拟用户数据
users = [
{"id": "s1001", "name": "张三"},
{"id": "s1002", "name": "李四"},
{"id": "s1003", "name": "王五"}
]
# 带路径参数和查询参数的GET请求
@app.get("/get_home/{hello}")
def get_home(hello: str, username: Union[str, None] = None, userpass: Union[str, None] = None):
return {"message": f"{hello}----{username}----{userpass}"}
# POST请求示例
@app.post("/post_home")
def post_home(username: str, userpass: str):
return users
if __name__ == '__main__':
uvicorn.run(app, host="127.0.0.1", port=8089)3.2 请求测试
import requests
import json
# 测试GET请求
res = requests.get("http://127.0.0.1:8089/get_home/abcabc")
data = json.loads(res.text)
print(data["message"])
# 测试POST请求
data = {"username": "admin", "userpass": "admin123"}
res = requests.post("http://127.0.0.1:8089/post_home", params=data)
print(json.loads(res.text))四、登录注册系统实现
4.1 API服务端
from fastapi import FastAPI
import uvicorn
app = FastAPI()
# 模拟用户数据库
user_db = {}
# 1. 注册接口
@app.post("/register")
def register(username: str, userpass: str):
user_db[username] = userpass
return {"message": "注册成功!"}
# 2. 登录接口
@app.get("/login")
def login(username: str, userpass: str):
if username in user_db and userpass == user_db[username]:
return {"status": 1, "message": "登录成功"}
else:
return {"status": 0, "message": "用户名或密码错误"}
# 3. 检查用户名是否存在
@app.post("/check_username")
def check_username(username: str):
if username in user_db:
return {"exists": True, "message": "用户名已存在"}
else:
return {"exists": False, "message": "用户名可用"}
if __name__ == '__main__':
uvicorn.run(app, host="127.0.0.1", port=8080)4.2 客户端调用
import requests
import json
def check_input(fn):
"""输入验证装饰器"""
def wrapper(username, password):
if username.strip() == "" or password.strip() == "":
return "用户名和密码不能为空！"
return fn(username, password)
return wrapper
@check_input
def register_user(username, password):
"""用户注册"""
# 先检查用户名是否存在
check_res = requests.post(
"http://127.0.0.1:8080/check_username",
params={"username": username}
)
check_data = json.loads(check_res.text)
if check_data["exists"]:
return "用户名已存在，请重新输入！"
# 执行注册
reg_res = requests.post(
"http://127.0.0.1:8080/register",
params={"username": username, "userpass": password}
)
return json.loads(reg_res.text)["message"]
@check_input
def login_user(username, password):
"""用户登录"""
res = requests.get(
f"http://127.0.0.1:8080/login",
params={"username": username, "userpass": password}
)
data = json.loads(res.text)
return data
def main():
while True:
print("\n===== 用户管理系统 =====")
print("1. 注册")
print("2. 登录")
print("3. 退出")
choice = input("请选择操作 (1/2/3): ")
if choice == "3":
print("感谢使用，再见！")
break
username = input("请输入用户名: ")
password = input("请输入密码: ")
if choice == "1":
result = register_user(username, password)
print(result)
elif choice == "2":
result = login_user(username, password)
if result["status"] == 1:
print(f"登录成功！欢迎 {username}")
# 这里可以添加登录后的其他操作
else:
print("登录失败:", result["message"])
if __name__ == "__main__":
main()五、员工管理系统完整实现
5.1 数据结构设计
from pydantic import BaseModel
from typing import Optional
class Employee(BaseModel):
"""员工数据模型"""
id: str # 员工编号
account: str # 账号
password: str # 密码
name: str # 姓名
age: Optional[int] = None # 年龄（可选）
phone: Optional[str] = None # 电话号码（可选）
address: Optional[str] = None # 家庭住址（可选）5.2 完整的员工管理系统API
from fastapi import FastAPI, HTTPException
import uvicorn
from typing import List, Optional
from pydantic import BaseModel
app = FastAPI()
# 员工数据模型
class Employee(BaseModel):
id: str
account: str
password: str
name: str
age: Optional[int] = None
phone: Optional[str] = None
address: Optional[str] = None
# 模拟数据库
employees_db = []
current_user = None # 当前登录用户
# 1. 登录API
@app.post("/login")
def login(account: str, password: str):
global current_user
for emp in employees_db:
if emp.account == account and emp.password == password:
current_user = emp
return {"message": "登录成功", "user": emp.name}
raise HTTPException(status_code=401, detail="账号或密码错误")
# 2. 注册API
@app.post("/register")
def register(employee: Employee):
# 检查账号是否已存在
for emp in employees_db:
if emp.account == employee.account:
raise HTTPException(status_code=400, detail="账号已存在")
# 检查编号是否已存在
for emp in employees_db:
if emp.id == employee.id:
raise HTTPException(status_code=400, detail="员工编号已存在")
employees_db.append(employee)
return {"message": "注册成功", "id": employee.id}
# 3. 根据账号查询员工
@app.get("/employee/{account}")
def get_employee_by_account(account: str):
for emp in employees_db:
if emp.account == account:
return emp
raise HTTPException(status_code=404, detail="员工不存在")
# 4. 查询所有员工（需要登录）
@app.get("/employees")
def get_all_employees():
if not current_user:
raise HTTPException(status_code=401, detail="请先登录")
return employees_db
# 5. 新增员工（需要登录）
@app.post("/employees")
def add_employee(employee: Employee):
if not current_user:
raise HTTPException(status_code=401, detail="请先登录")
# 验证必填字段
if not all([employee.id, employee.account, employee.password, employee.name]):
raise HTTPException(status_code=400, detail="编号、账号、密码、姓名为必填项")
# 检查编号和账号是否重复
for emp in employees_db:
if emp.id == employee.id:
raise HTTPException(status_code=400, detail="员工编号已存在")
if emp.account == employee.account:
raise HTTPException(status_code=400, detail="员工账号已存在")
employees_db.append(employee)
return {"message": "新增成功", "employee": employee}
# 6. 删除员工（需要登录）
@app.delete("/employees/{employee_id}")
def delete_employee(employee_id: str):
if not current_user:
raise HTTPException(status_code=401, detail="请先登录")
for i, emp in enumerate(employees_db):
if emp.id == employee_id:
deleted = employees_db.pop(i)
return {"message": "删除成功", "deleted_employee": deleted}
raise HTTPException(status_code=404, detail="员工不存在")
# 7. 修改员工信息（需要登录）
@app.put("/employees/{employee_id}")
def update_employee(employee_id: str, age: Optional[int] = None,
phone: Optional[str] = None, address: Optional[str] = None):
if not current_user:
raise HTTPException(status_code=401, detail="请先登录")
for emp in employees_db:
if emp.id == employee_id:
if age is not None:
emp.age = age
if phone is not None:
emp.phone = phone
if address is not None:
emp.address = address
return {"message": "修改成功", "updated_employee": emp}
raise HTTPException(status_code=404, detail="员工不存在")
if __name__ == '__main__':
uvicorn.run(app, host="127.0.0.1", port=8000)六、路由分发与API组织
6.1 使用APIRouter模块化路由
# api/employees.py
from fastapi import APIRouter, HTTPException
from typing import List
from models import Employee
router = APIRouter()
# 员工相关的路由
@router.get("/", response_model=List[Employee])
async def get_all_employees():
return []
@router.post("/", response_model=Employee)
async def create_employee(employee: Employee):
return employee
@router.put("/{employee_id}", response_model=Employee)
async def update_employee(employee_id: str, employee: Employee):
return employee
@router.delete("/{employee_id}")
async def delete_employee(employee_id: str):
return {"message": "删除成功"}6.2 主应用集成
# main.py
from fastapi import FastAPI
from api import employees, auth
app = FastAPI(title="员工管理系统API")
# 包含路由
app.include_router(auth.router, prefix="/auth", tags=["认证"])
app.include_router(employees.router, prefix="/employees", tags=["员工管理"])七、使用FastAPI的交互式文档
启动应用后，FastAPI会自动生成交互式API文档：
-
Swagger UI文档：访问
http://127.0.0.1:8000/docs -
ReDoc文档：访问
http://127.0.0.1:8000/redoc
在文档中你可以：
-
查看所有API端点
-
查看请求/响应模型
-
直接测试API接口
-
查看HTTP状态码和错误信息
八、总结
本文详细介绍了如何使用FastAPI构建一个完整的员工管理系统，涵盖了：
-
FastAPI基础：安装、配置、基本路由
-
核心概念：HTTP请求、序列化、类型提示
-
认证系统：登录、注册、权限控制
-
CRUD操作：增删改查员工信息
-
路由组织：模块化、APIRouter使用
-
API文档：自动生成交互式文档
腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。
更多推荐
14
0- 0
已为社区贡献1条内容
所有评论(0)