FastAPI 学习（二）

FastAPI 学习（二）

概要：本篇内容主要是对上篇文章的补充，主要的内容有FastAPI 参数接收、日志记录、用户文件上传、Demo 示例，后续将对这部分内容展开叙述；
1. FastAPI 参数接收

FastAPI 接收参数的方式与django和flask略有不同，下面将依次介绍用到的方法;
1.1 路径参数

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : fast_app.py
6. @Time    : 2025/6/2 22:47
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 学习使用 FastAPI 接收用户的参数的信息;
10. """
11. import uvicorn
12. from fastapi import FastAPI
14. app = FastAPI()
17. @app.get("/items/{item_id}")
18. async def read_item(item_id: int):
19.     """
20.     通过 URL 的路径获取到参数的信息; 在路由装饰器中使用路径参数，通过函数参数接收。
21.     Args:
22.         item_id:
24.     Returns:
26.     """
27.     print(item_id)  # >>> 1
28.     return {"item_id": item_id}
31. if __name__ == '__main__':
32.     uvicorn.run(app, host='127.0.0.1', port=8000)

复制代码

说明：item_id 是路径参数，类型为 int，FastAPI 会自动解析路径中的参数并转换为指定类型。如果路径中没有匹配的参数，FastAPI 会返回 404 错误。
1.2 url 参数

使用？拼接在 url 末端的参数的信息；

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : fast_app.py
6. @Time    : 2025/6/2 22:47
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 学习使用 FastAPI 接收用户的参数的信息;
10. """
12. import uvicorn
13. from fastapi import FastAPI, Query
15. app = FastAPI()
18. @app.get("/index")
19. async def index(q: str = Query(None, max_length=50)):
20.     """
21.     通过 URL拼接在函数的尾部; 在函数参数中定义查询参数，使用 Query 类型标注。
22.     Args:
23.         q:
25.     Returns:
27.     """
28.     results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
29.     if q:
30.         results.update({"q": q})
31.     return results
34. if __name__ == '__main__':
35.     uvicorn.run(app, host='127.0.0.1', port=8000)

复制代码

说明:

• q 是查询参数，默认值为 None，max_length=50 表示查询参数的最大长度为 50。
  
• 如果没有传递查询参数 q，则使用默认值 None。
  

1.3 请求体接收

这种方式通常是使用 Pydantic 模型定义请求体的结构，通过函数参数接收

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : fast_app.py
6. @Time    : 2025/6/2 22:47
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 学习使用 FastAPI 接收用户的参数的信息;
10. """
12. import uvicorn
13. from fastapi import FastAPI, Query
15. from pydantic import BaseModel
17. app = FastAPI()
20. class Item(BaseModel):
21.     """ 定义 Pydantic 模型参数;
22.     """
23.     name: str
24.     description: str = None
25.     price: float
26.     tax: float = None
29. @app.post("/items/")
30. async def create_item(item: Item):
31.     print(item)
32.     return {"item": item}
35. if __name__ == '__main__':
36.     uvicorn.run(app, host='127.0.0.1', port=8000)

复制代码

说明：

• Item是一个 Pydantic 模型，定义了请求体结构；
  
• FastAPI 会自动解析请求体中的JSON数据，并将其转换成为Item对象。
  
• 如果请求体中的数据不符合Item模型的定义，FastAPI 会返回 422 错误。
  

1.4 表单接收

定义方式：使用 Form 类型标注接收表单数据，主要用来接收前端 Form 表单校验形式的数据；

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : learn_arg.py
6. @Time    : 2025/6/3 22:42
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 学习 FastAPI 使用的参数解析形式;
10. """
11. import uvicorn
12. from fastapi import FastAPI, Form
14. app = FastAPI()
17. @app.post("/login")
18. def login(username: str = Form(...), password: str = Form(...)):
19.     return {"username": username, "password": password}
22. if __name__ == '__main__':
23.     uvicorn.run(app, host="127.0.0.1", port=8000)

复制代码

• username 和 password 是表单数据，使用 Form 类型标注。
  
• ... 表示该参数是必填项。
  
• FastAPI 会自动解析表单数据并将其传递给函数参数。
  

1.5 文件上传

定义方式：使用 File 和 UploadFile 类型标注接收文件。

1. import uvicorn
2. from fastapi import FastAPI, File, UploadFile
4. app = FastAPI()
7. # @app.post("/login")
8. # def login(username: str = Form(...), password: str = Form(...)):
9. #     return {"username": username, "password": password}
11. @app.post("/files/")
12. async def create_user(file: bytes = File(...)):
13.     # print(len(file.size))
14.     return {"file": len(file)}
17. @app.post("/uploadfile/")
18. async def create_upload_file(file: UploadFile = File(...)):
19.     return {"filename": file.filename}
22. if __name__ == '__main__':
23.     uvicorn.run(app, host="127.0.0.1", port=8000)

复制代码

• file 是上传的文件，使用 File 类型标注。
  
• UploadFile 是一个更高级的文件对象，包含文件名、内容类型等信息。
  
• FastAPI 会自动解析上传的文件并将其传递给函数参数。
  

1.6 请求头接收

在 FastAPI 中可以使用 Header 类型标注请求接收头；

1. import uvicorn
2. from fastapi import FastAPI, File, UploadFile, Header
4. app = FastAPI()
6. @app.get("/items/")
7. async def read_items(user_agent: str = Header(None)):
8.     return {"ua": user_agent}
11. if __name__ == '__main__':
12.     uvicorn.run(app, host="127.0.0.1", port=8000)

复制代码

1.7 依赖项接收

使用 Depneds 的方式来接收参数的信息；

1. import uvicorn
2. from fastapi import FastAPI, File, UploadFile, Header, Depends
4. from sqlalchemy import create_engine
5. from sqlalchemy.orm import sessionmaker, Session
7. app = FastAPI()
10. engine = create_engine("my.db")
11. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
14. def get_db():
15.     db = SessionLocal()
16.     try:
17.         yield db
18.     except Exception as e:
19.         print(e)
20.         db.rollback()
21.     finally:
22.         db.close()
25. @app.get("/items/")
26. async def read_items(name: str, db: Session = Depends(get_db)):
27.     data = db.query()  # 此处进行数据库的操作;
28.     return data
31. if __name__ == '__main__':
32.     uvicorn.run(app, host="127.0.0.1", port=8000)

复制代码

接收依赖项, 对数据库进行操作；

1. from fastapi import FastAPI, Depends, Header
3. app = FastAPI()
5. async def verify_token(x_token: str = Header(...)):
6.     if x_token != "fake-super-secret-token":
7.         raise HTTPException(status_code=400, detail="X-Token header invalid")
9. @app.get("/items/", dependencies=[Depends(verify_token)])
10. async def read_items():
11.     return [{"item": "Foo"}, {"item": "Bar"}]

复制代码

• verify_token 是一个依赖项函数，验证请求头中的 X-Token 字段。
  
• dependencies=[Depends(verify_token)] 表示在调用路由函数之前，先调用 verify_token 函数。
  

额外补充:

多次依赖：FastAPI 的依赖注入机制会确保同一个依赖项在一次请求处理中只被调用一次，即使它在多个地方被引用。

1.8 背景任务接收

使用 BackgroundTasks 类型标注接收背景任务；

1. from fastapi import FastAPI, BackgroundTasks
3. app = FastAPI()
5. def write_notification(email: str, message=""):
6.     with open("log.txt", mode="w") as email_file:
7.         content = f"notification for {email}: {message}"
8.         email_file.write(content)
10. @app.post("/send-notification/{email}")
11. async def send_notification(email: str, background_tasks: BackgroundTasks):
12.     background_tasks.add_task(write_notification, email, message="some notification")
13.     return {"message": "Notification sent in the background"}

复制代码

• background_tasks 是一个背景任务对象，使用 BackgroundTasks 类型标注。
  
• background_tasks.add_task 方法可以添加背景任务，任务会在响应返回后执行。
  
• 这种场景适合做任务执行的开启，挂起任务到后台的时候使用;
  

1.9 请求对象接收

通过 Request 对象的参数进行接收;

这种方式，与 Flask 和 django 的获取参数的形式比较相似；

1. from fastapi import FastAPI, Request
3. app = FastAPI()
5. @app.get("/items/")
6. async def read_items(request: Request):
7.     return {"client_host": request.client.host}

复制代码

这种方式获取的时候通常获取在请求头中的一些数据，在辅助验证的时候有着不错的效果；request 是一个请求对象，包含请求的完整信息，如客户端 IP 地址、请求头等。
通过以上多种方式，FastAPI 提供了灵活且强大的参数接收机制，可以根据实际需求选择合适的方式接收用户参数。
2. FastAPI 日志记录

额外补充：日志记录中的 Logger 与 Handler；
Logger 是日志系统的入口，负责产生日志记录，并可以设置日志级别来控制整个日志记录流程的初始过滤。Handler 则负责将日志记录发送到特定的目标位置，如文件、控制台、邮件等。一个 Logger 对象可以通过addHandler方法添加0到多个 Handler，每个 Handler 可以定义不同的日志级别，以实现日志分级过滤和显示‌；

• Logger
  
  
  
  • ‌创建和管理‌：Logger 负责创建或获取用于发送日志记录消息的对象。用户可以通过logging.getLogger()创建或获取一个Logger对象‌。
    
  • 日志级别：Logger 可以设置日志级别，控制日志记录的初始过滤。只有满足 Logger 级别要求的日志记录才会被处理和输出‌。
    
  • 日志记录发送：Logger 对象用于记录日志消息，用户通过 Logger 对象发送日志记录‌。
    
  
  
• Handler
  
  
  
  • 目标指定：Handler 指定如何处理和输出日志消息，可以将日志消息写入文件、控制台、网络等不同的目标‌。常见的 Handler子类包括StreamHandler（输出到控制台）、FileHandler（输出到文件）和SMTPHandler（输出到邮件）等‌。
    
  • 日志级别设置：Handler 可以设置自己的日志级别，只有满足 Handler 级别要求的日志记录才会被处理和输出。Logger 和 Handler 的级别都需要满足，日志记录才会被处理和输出‌。
    
  • 格式化：Handler 还可以设置日志格式化器（Formatter），定义日志记录的输出格式，如时间戳、日志级别、日志名称和消息内容等‌。
    
  
  

2.1 日志输出到文件

1. import logging
2. from fastapi import FastAPI
4. # 配置日志
5. logger = logging.getLogger(__name__)
6. logger.setLevel(logging.INFO)
8. # 创建文件处理器
9. file_handler = logging.FileHandler('app.log')
10. file_handler.setLevel(logging.INFO)
12. # 创建控制台处理器
13. console_handler = logging.StreamHandler()
14. console_handler.setLevel(logging.INFO)
16. # 创建日志格式
17. formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
18. file_handler.setFormatter(formatter)
19. console_handler.setFormatter(formatter)  # 设置终端输出的日志的格式;
23. # 将文件处理器添加到日志器
24. logger.addHandler(file_handler)
25. logger.addHandler(console_handler)  # 一个 logger 可以添加多个 handler;
27. app = FastAPI()
29. @app.get("/")
30. async def read_root():
31.     logger.info("Handling root endpoint")
32.     return {"message": "Hello World"}
34. """
35. 在这个例子中：
36.         FileHandler 用于将日志写入文件 app.log。
37.         Formatter 用于定义日志的格式。
38.         addHandler 方法将文件处理器添加到日志器。
39. """

复制代码

以上是传统的日志记录的方式，可以将我们的日志同时输出到终端和文件中，但是记录的仅仅是我们开发过程中自己定义的日志记录的信息，如果我们想要将终端的日志也输出到文件中，应该怎么操作呢?

基于启动的 uvicorn 实现；

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : base.py
6. @Time    : 2025/6/4 23:10
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 基于 uvicorn 实现日志的记录;
10. """
11. import logging
12. from logging.handlers import RotatingFileHandler
13. from fastapi import FastAPI
14. import uvicorn
16. # 创建 FastAPI 应用
17. app = FastAPI()
19. # 配置日志
21. LOGGING_CONFIG = {
22.     "version": 1,
23.     "disable_existing_loggers": False,
24.     "formatters": {
25.         "default": {
26.             "()": "uvicorn.logging.DefaultFormatter",
27.             "fmt": "%(levelprefix)s %(message)s",
28.             "use_colors": None,
29.         },
30.         "access": {
31.             "()": "uvicorn.logging.AccessFormatter",
32.             "fmt": '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
33.         },
34.     },
35.     "handlers": {
36.         "default": {
37.             "formatter": "default",
38.             "class": "logging.handlers.TimedRotatingFileHandler",
39.             "filename": "./fastapi.log"
40.         },
41.         "access": {
42.             "formatter": "access",
43.             "class": "logging.handlers.TimedRotatingFileHandler",
44.             "filename": "./fastapi.log"
46.         },
47.     },
48.     "loggers": {
49.         "": {"handlers": ["default"], "level": "INFO"},
50.         "uvicorn.error": {"level": "INFO"},
51.         "uvicorn.access": {"handlers": ["access"], "level": "INFO", "propagate": False},
52.     },
53. }
56. @app.get("/")
57. async def read_root():
58.     return {"message": "Hello, World!"}
61. if __name__ == "__main__":
62.     uvicorn.run(app, host="127.0.0.1", port=8000, log_config=LOGGING_CONFIG)

复制代码

2.2 中间件实现

基于 FastAPI 中间件实现日志的记录；

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : base.py
6. @Time    : 2025/6/4 23:10
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 基于 FastAPI 中间件实现日志的记录；
10. """
11. from fastapi import FastAPI, Request
12. from loguru import logger
13. import uvicorn
15. # 创建 FastAPI 应用
16. app = FastAPI()
18. # 配置日志
19. logger.add("./request.log", rotation="100 MB")  # 自定义日志文件路径和大小限制
21. @app.middleware("http")
22. async def log_requests(request: Request, call_next):
23.     """
24.     中间件：记录请求日志
25.     """
26.     logger.info(f"Request: {request.method} {request.url}")
27.     response = await call_next(request)
28.     logger.info(f"Response: {response.status_code}")
29.     return response
31. @app.get("/")
32. async def read_root():
33.     return {"message": "Hello, World!"}
35. if __name__ == "__main__":
36.     uvicorn.run(app, host="127.0.0.1", port=8000)

复制代码

2.2 审计日志的记录

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : base.py
6. @Time    : 2025/6/4 23:10
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    :
10. """
12. import time
14. import uvicorn
15. from loguru import logger
16. from fastapi import FastAPI, Request
17. from starlette.middleware.base import BaseHTTPMiddleware
19. # 配置日志
20. logger.add("./request.log", rotation="100 MB")  # 自定义日志文件路径和大小限制
21. # 创建 FastAPI 应用
22. app = FastAPI()
25. class AuditLogMiddleware(BaseHTTPMiddleware):
26.     async def dispatch(self, request: Request, call_next):
27.         # 在请求到达路由处理函数之前执行的代码
28.         start_time = time.time()
29.         logger.info(f"Request received: {request.method} {request.url}")
31.         # 调用下一个中间件或路由处理函数
32.         response = await call_next(request)
34.         # 在响应返回客户端之前执行的代码
35.         process_time = time.time() - start_time
36.         logger.info(
37.             f"Request processed: {request.method} {request.url} - Status: {response.status_code} - Time: {process_time:.2f}s")
39.         return response
42. # 添加审计日志中间件
43. app.add_middleware(AuditLogMiddleware)
46. @app.get("/")
47. async def read_root():
48.     return {"message": "Hello World"}
51. if __name__ == "__main__":
52.     uvicorn.run(app, host="127.0.0.1", port=8000)

复制代码

基于中间件的方式的时候可以记录到一些关于客户请求端的信息到日志中；
2.3 装饰器方式

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : base.py
6. @Time    : 2025/6/4 23:10
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 装饰器实现日志的记录
10. """
12. import logging
13. from typing import Callable, Any
14. from functools import wraps
16. import uvicorn
17. from fastapi import FastAPI
19. # 配置日志
20. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
21. logger = logging.getLogger(__name__)
24. def log_decorator(module_name: str, user_description: str):
25.     """
26.     日志装饰器，记录模块名称和用户描述
27.     :param module_name: 模块名称
28.     :param user_description: 用户描述
29.     """
31.     def decorator(func: Callable) -> Callable:
32.         @wraps(func)
33.         async def wrapper(*args: Any, **kwargs: Any) -> Any:
34.             # 记录请求前的日志, 调用原有的参数
35.             logger.info(f"Module: {module_name} - Description: {user_description} - Request: {func.__name__}")
37.             # 调用原始函数
38.             result = await func(*args, **kwargs)
40.             # 记录请求后的日志
41.             logger.info(
42.                 f"Module: {module_name} - Description: {user_description} - Response: {func.__name__} - Result: {result}")
44.             return result
46.         return wrapper
48.     return decorator
51. app = FastAPI()
54. @app.get("/")
55. @log_decorator(module_name="主模块", user_description="访问根路径")
56. async def read_root():
57.     return {"message": "Hello World"}
60. @app.get("/items/{item_id}")
61. @log_decorator(module_name="项目模块", user_description="获取项目详情")
62. async def read_item(item_id: int):
63.     return {"item_id": item_id, "message": "Item retrieved"}
66. if __name__ == '__main__':
67.     uvicorn.run(app, host="127.0.0.1", port=8000)

复制代码

此时日志已经输出到终端上面，并且包含了自定义的信息；如果需要用户的日志输出到别的地方如：消息队列、数据库、文件、API 接口的时候，用户可以自定义 Handdler 并且配置到 logger 中即可实现；
3. FastAPI 登录验证

登录验证的流程为常见的开发示例，可以开发完成之后备用为已有的封装服务，方便后续的开发工作的完善;
3.1 设定配置文件

本步骤的目的是为了设定存储数据的位置，以及用到的组件的信息;

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : settings.py
6. @Time    : 2025/6/5 23:10
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 配置文件
10. """
11. import os
13. from pydantic import Field
14. from pydantic_settings import BaseSettings, SettingsConfigDict
15. from sqlalchemy import create_engine
16. from sqlalchemy.orm import sessionmaker
19. class Settings(object):
20.     """原生的配置 Python 文件的写法;
21.     """
23.     # 使用 SQL lite 作为数据库;
24.     DATABASE_URL = "sqlite:///data.db"
25.     engine = create_engine(DATABASE_URL, echo=True)
26.     SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
28.     # SECRET_KEY
29.     ALGORITHM = "HS256"
30.     SECRET_KEY = "fshduifhue"
31.     ACCESS_TOKEN_EXPIRE_MINUTES = 30
34. class SettingsEnv(BaseSettings):
35.     """ 使用 .env 文件加载配置文件;
36.     """
38.     # SettingsConfigDict 如果环境变量存在相同名称的变量, 则覆盖文件内的, 否则使用文件内的键值对的变量形式;
39.     model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="forbid")
41.     DB_NAME: str = Field("data.db", description="数据库名称")
43.     REDIS_HOST: str = Field("127.0.0.1", description="redis 服务地址")
44.     REDIS_PORT: int = Field(6379, description="redis 端口号地址")
45.     REDIS_USER: str = Field("", description="redis 用户名")
46.     REDIS_PASSWORD: str = Field("", description="redis 密码")
47.     SECRET_KEY: str = Field("", description="FastAPI 加密秘钥")
50. class DatabaseSession:
51.     """定义上下文处理连接;
52.     """
54.     def __init__(self):
55.         self.db = Settings.SessionLocal()
57.     def __enter__(self):
58.         return self.db
60.     def __exit__(self, exc_type, exc_val, exc_tb):
61.         self.db.close()

复制代码

以上的两种模式可以自行选择，如果使用第二种，可以自行根据字段编写.env文件；
3.2 设置模型

本步骤的目的是用来设定登录用户中常见的数据库的模型以及 Pydantic 模型，方便后续的业务开展；

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : models.py
6. @Time    : 2025/6/5 23:07
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 定义数据库模型, 创建数据库表, 登录过程中创建一张表
10. """
12. from sqlalchemy import , Column, Integer, String
13. from sqlalchemy.ext.declarative import declarative_base
14. from user_web.settings import Settings
16. Base = declarative_base()
19. class User(Base):
20.     """用户表;
21.     """
23.     __tablename__ = "user"
24.     id = Column(Integer, primary_key=True, index=True, comment="用户ID")
25.     username = Column(String, unique=True, index=True, comment="用户名")
26.     hash_password = Column(String, comment="hash 密码")
29. if __name__ == '__main__':
30.     # 创建数据库;
31.     Base.metadata.create_all(bind=Settings.engine)

复制代码

如需其他字段可以自行添加，本博客只做最简单的记录，不做其他的处理，后续补充可能会更新到 Git 上，可通过文章结尾的地址进行访问。

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : schemas.py
6. @Time    : 2025/6/5 23:42
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 定义 Pydantic 模型,辅助校验;
10. """
12. from pydantic import BaseModel, Field
15. class User(BaseModel):
16.     id: int
17.     username: str

复制代码

3.3 ORM 操作封装

实现数据库常见操作的封装，本次操作数据库使用 SQLAlchemy，封装好的工具也可以在其他地方进行使用。

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : base_sqlalchemy_utils.py
6. @Time    : 2025/6/30 14:29
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 数据库操作的基类，定义些通用的数据操作方法
10. """
12. import math
14. from sqlalchemy import desc, asc
15. from sqlalchemy.orm.query import Query
16. from sqlalchemy.orm.session import Session
17. from sqlalchemy_utils import Choice
20. class BaseDBOperateModel(object):
21.     @classmethod
22.     def get_list(cls, db_query: Query, filters: set, order: str = "-id", offset: int = 0, limit: int = 15) -> dict:
23.         """分页查询数据
25.         Args:
26.             db_query: Query;数据库session绑定了查询数据库模型的查询类
27.             filters: set;过滤条件
28.             order: str;排序规则，例："+id,-create_time"
29.             offset: int;偏移量
30.             limit: int;取多少条
32.         Returns:
33.             dict;数据结果
34.         """
35.         # noinspection All
36.         count_number = db_query.filter(*filters).count()
37.         result = {
38.             "page": {
39.                 "count": count_number,
40.                 "total_page": cls.get_page_number(count_number, limit),
41.                 "current_page": offset
42.             },
43.         }
44.         filter_result = list()
45.         if offset != 0:
46.             offset = (offset - 1) * limit
47.         if result["page"]["count"] > 0:
48.             # 禁用全部检查
49.             # noinspection All
50.             filter_query = db_query.filter(*filters)
51.             order_rules = cls.order_transfer(order)
52.             # noinspection All
53.             filter_result = filter_query.order_by(*order_rules).offset(offset).limit(limit).all()
54.         result["list"] = [cls.to_dict(c) for c in filter_result]
55.         return result
57.     @classmethod
58.     def get_all(cls, db_query: Query, filters: set, order: str = "-id", limit: int = 0) -> list:
59.         """获取所有符合条件的数据（有问题，暂不使用）
61.         Args:
62.             db_query: Query;数据库session绑定了查询数据库模型的查询类
63.             filters: set;过滤条件
64.             order: str;排序规则，例："+id,-create_time"
65.             limit: int;取多少条
67.         Returns:
68.             list;多条查询数据结果
69.         """
70.         if not filters:
71.             result = db_query
72.         else:
73.             # noinspection All
74.             result = db_query.filter(*filters)
75.         order_rules = cls.order_transfer(order)
76.         result.order_by(*order_rules)
77.         if limit != 0:
78.             # noinspection All
79.             result = result.limit(limit)
80.         # noinspection All
81.         result = result.all()
82.         result = [cls.to_dict(c) for c in result]
83.         return result
85.     @classmethod
86.     def get_one(cls, db_query: Query, filters: set, order: str = "-id") -> dict:
87.         """获取一条符合条件的数据
89.         Args:
90.             db_query: Query;数据库session绑定了查询数据库模型的查询类
91.             filters: set;过滤条件
92.             order: str;排序规则，例："+id,-create_time"
94.         Returns:
95.             dict;单条查询数据结果
96.         """
97.         # noinspection All
98.         result = db_query.filter(*filters)
99.         order_rules = cls.order_transfer(order)
100.         # noinspection All
101.         result = result.order_by(*order_rules).first()
102.         if result is None:
103.             return {}
104.         result = cls.to_dict(result)
105.         return result
107.     @staticmethod
108.     def insert(db_session: Session, model, data: dict) -> int:
109.         """插入一条数据
111.         Args:
112.             db_session: Session;数据库会话连接
113.             model: BaseModel;数据模型类对象
114.             data: set;插入数据
116.         Returns:
117.             int;插入成功后返回的id编号
118.         """
119.         users = model(**data)
120.         db_session.add(users)
121.         db_session.flush()
122.         return users.id
124.     @staticmethod
125.     def insert_all(db_session: Session, model, data: list) -> bool:
126.         """插入多条数据
128.         Args:
129.             db_session: Session;数据库会话连接
130.             model: BaseModel;数据模型类对象
131.             data: set;插入数据
133.         Returns:
134.             bool;插入多条数据成功返回True
135.         """
136.         users = list()
137.         for user_info in data:
138.             users.append(model(**user_info))
139.         db_session.add_all(users)
140.         db_session.commit()
141.         return True
143.     @staticmethod
144.     def update(db_query: Query, data: dict, filters: set) -> int:
145.         """修改符合条件的数据
147.         Args:
148.             db_query: Query;数据库session绑定了查询数据库模型的查询类
149.             data: dict;插入数据
150.             filters: set;过滤条件
152.         Returns:
153.             int;修改数据成功返回的行数
154.         """
155.         # noinspection All
156.         return db_query.filter(*filters).update(data, synchronize_session=False)
158.     @staticmethod
159.     def delete(db_query: Query, filters: set) -> int:
160.         """删除符合条件的数据
162.         Args:
163.             db_query: Query;数据库session绑定了查询数据库模型的查询类
164.             filters: set;过滤条件
166.         Returns:
167.              int;修改数据成功返回的行数
168.         """
169.         # noinspection All
170.         return db_query.filter(*filters).delete(synchronize_session=False)
172.     @staticmethod
173.     def get_count(db_query: Query, filters: set, field=False) -> int:
174.         """获取符合条件的数据数量
176.         Args:
177.             db_query: Query;数据库session绑定了查询数据库模型的查询类
178.             filters: set;过滤条件
179.             field: bool;是否指定字段计数
181.         Returns:
182.             int;
183.         """
184.         if field:
185.             # noinspection All
186.             return db_query.filter(*filters).scalar()
187.         else:
188.             # noinspection All
189.             return db_query.filter(*filters).count()
191.     @staticmethod
192.     def get_page_number(count: int, page_size: int) -> int:
193.         """获取总页数
195.         Args:
196.             count: int;数据总数
197.             page_size: int;分页大小
199.         Returns:
200.             int;总页数
201.         """
202.         page_size = abs(page_size)
203.         if page_size != 0:
204.             total_page = math.ceil(count / page_size)
205.         else:
206.             total_page = math.ceil(count / 5)
207.         return total_page
209.     @staticmethod
210.     def order_transfer(order: str):
211.         """order排序规则转换
213.         Args:
214.             order: str;排序规则，例："+id,-create_time"
216.         Returns:
217.             list;排序规则列表
218.         """
219.         order_array = order.split(",")
220.         order_rules = list()
221.         for item in order_array:
222.             sort_rule = item[0]
223.             if sort_rule == "-":
224.                 order_rules.append(desc(item[1:]))
225.             else:
226.                 order_rules.append(asc(item[1:]))
227.         return order_rules
229.     @staticmethod
230.     def to_dict(model_obj):
231.         if not hasattr(model_obj, "_fields"):
232.             only = model_obj.__table__.columns
233.             result = {field.name: (
234.                 getattr(model_obj, field.name).value if isinstance(getattr(model_obj, field.name), Choice) else getattr(
235.                     model_obj, field.name)) for field in only}
236.         else:
237.             only = model_obj.keys()
238.             result = {field: (
239.                 getattr(model_obj, field).value if isinstance(getattr(model_obj, field), Choice) else getattr(
240.                     model_obj, field)) for field in only}
241.         return result

复制代码

上述的操作，没有设置数据库连接的获取与关闭，因为获取连接的方式封装在了配置文件内；
3.4 业务 API 的开发

API 常见业务的封装；

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : auth_logic.py
6. @Time    : 2025/6/5 23:17
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 认证的主要逻辑函数;
10. """
11. from typing import Optional, Dict
12. from datetime import datetime, timedelta
14. from jose import jwt, JWTError
15. from passlib.context import CryptContext
16. from fastapi import Depends, HTTPException
17. from fastapi.security import OAuth2PasswordBearer
19. from models import User
20. from base_sqlalchemy_utils import BaseDBOperateModel
21. from settings import Settings, DatabaseSession
23. # 定义密码上下哈希文
24. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
26. # JWT 进行配置
27. SECRET_KEY = Settings.SECRET_KEY
29. ACCESS_TOKEN_EXPIRE_MINUTES = 30
31. # 设定密码模式
32. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
35. def verify_password(plain_password, hashed_password):
36.     """
37.     校验原密码与 hash 存储的值;
38.     :param plain_password:
39.     :param hashed_password:
40.     :return:
41.     """
42.     return pwd_context.verify(plain_password, hashed_password)
45. def get_password_hash(password):
46.     """
47.     获取密码的 Hash 值;
48.     :param password:
49.     :return:
50.     """
51.     return pwd_context.hash(password)
54. def create_access_token(data: Dict, expires_delta: Optional[timedelta] = None):
55.     """
56.     创建 token 函数;
57.     :param data:
58.     :param expires_delta:
59.     :return:
60.     """
61.     to_encode = data.copy()
62.     expire = datetime.utcnow() + timedelta(minutes=Settings.ACCESS_TOKEN_EXPIRE_MINUTES)
63.     to_encode.update({"exp": expire})
64.     encoded_jwt = jwt.encode(to_encode, Settings.SECRET_KEY, algorithm=Settings.ALGORITHM)
65.     return encoded_jwt
68. async def get_current_user(token: str = Depends(oauth2_scheme)) -> Dict:
69.     """
70.     实现 Token 对用户的解析;
71.     :param token:
72.     :return:
73.     """
74.     credentials_exception = HTTPException(
75.         status_code=401, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}
76.     )
77.     try:
78.         payload = jwt.decode(token, Settings.SECRET_KEY, algorithms=[Settings.ALGORITHM])
79.         username: str = payload.get("user_name")
80.         if username is None:
81.             raise credentials_exception
82.     except JWTError:
83.         raise credentials_exception
84.     with DatabaseSession() as db_session:
85.         user = BaseDBOperateModel.get_one(db_session.query(User), {User.username == username})
86.         if not user:
87.             raise credentials_exception
88.     return user

复制代码

核心的加密逻辑如上。

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : user_api.py
6. @Time    : 2025/6/5 23:29
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    : 使用用户的路由配置;
10. """
11. from typing import Dict
12. from datetime import timedelta
14. from fastapi import APIRouter, Depends, HTTPException
16. from auth_logic import create_access_token, get_current_user
17. from models import User
18. from schemas import UserCreate
19. from settings import DatabaseSession, Settings
20. from auth_logic import get_password_hash, verify_password
21. from base_sqlalchemy_utils import BaseDBOperateModel
23. user_router = APIRouter(prefix="/user", tags=["user"])
26. @user_router.post("/register")
27. async def register(user_data: UserCreate):
28.     """
29.     用户注册;
30.     user_data: UserCreate; pydantic.BaseModel; 设置参数的基本校验;
31.     :return:
32.     """
33.     # 1. 首先检查用户名是否存在, 存在返回异常的信息
34.     with DatabaseSession() as db_session:
35.         user = BaseDBOperateModel.get_one(db_session.query(User), filters={User.username == user_data.username})
36.         if user:
37.             raise HTTPException(status_code=400, detail="username already exists.")
38.         # 2. 用户名不存在, hash 密码,创建用户;
39.         hashed_password = get_password_hash(user_data.password)
40.         user = BaseDBOperateModel.insert(db_session, User,
41.                                          {"username": user_data.username, "hash_password": hashed_password})
42.         db_session.commit()
43.     return {"code": 200, "message": "success", "data": {"id": user, "username": user_data.username}}
46. @user_router.post("/login-token")
47. async def login(user_data: UserCreate):
48.     """
49.     登录验证;
50.     :param user_data:
51.     :return:
52.     """
53.     # 1. 对参数的进行第一步的叫校验;
54.     user_name = user_data.username
55.     password = user_data.password
56.     with DatabaseSession() as db_session:
57.         user_obj = BaseDBOperateModel.get_one(db_session.query(User), filters={User.username == user_name})
58.         status_pwd = verify_password(password, user_obj.get("hash_password"))  # 校验字段密码的信息;
59.         if not status_pwd:
60.             raise HTTPException(status_code=400, detail="invalid username or password")
61.         # 通过密码的校验, 进行 token 的创建;
62.         access_token = create_access_token(data={"user_id": user_obj.get("id"), "user_name": user_name},
63.                                            expires_delta=timedelta(minutes=Settings.ACCESS_TOKEN_EXPIRE_MINUTES))
64.     return {"status": 200, "message": "success", "data": {"access_token": access_token, "token_type": "bearer"}}
66. @user_router.get("/home")
67. async def home(current_user: Dict = Depends(get_current_user)):
68.     """
69.     登录成功之后的跳转路由, 基于依赖注入做认证;
70.     :param current_user: User,
71.     :return:
72.     """
73.     print(current_user.get("username"), "登录成功了")
74.     return {"code": 200, "message": "Hello success message.", "data": {"id": current_user.get("id")}}

复制代码

API 部分的代码如上显示；
3.5 程序启动

提供启动方式，uvicorn，其他 corn 的启动方式可以百度参考。

1. # !/usr/bin/env python
2. # -*-coding:utf-8 -*-
4. """
5. @File    : __init__.py.py
6. @Time    : 2025/6/5 23:03
7. @Author  : zi qing bao jian
8. @Version : 1.0
9. @Desc    :
10. """
11. from fastapi import FastAPI
12. from .user_api import user_router
15. def create_app():
16.     app = FastAPI()
17.     app.include_router(user_router, prefix="/api")
18.     return app

复制代码

编写启动文件。

1. import uvicorn
2. from user_web import create_app  # 挂载路由的信息写入 __init__.py 的文件中, 对外部隐藏 app 细节;
4. app = create_app()
6. if __name__ == '__main__':
7.     uvicorn.run(app, host="127.0.0.1", port=8000)

复制代码

继续努力，终成大器。

来源：豆瓜网用户自行投稿发布，如果侵权，请联系站长删除