Python异步开发实践：FastAPI与Tortoise-ORM的高效协同工作指南

一、前言

在 Python Web 开发领域，框架的选择始终与场景深度绑定：Flask 的 “微内核 + 插件” 模式赋予开发者自由组装功能的灵活性，却在大型项目中暴露组件碎片化与异步支持不足的问题；Django 的 “全栈一站式” 架构降低开发门槛，但其同步执行机制在高并发 API 服务中逐渐力不从心；而 FastAPI 以异步原生、类型安全和高性能设计，重新定义了微服务与 I/O 密集型场景的开发体验。

数据库交互层同样存在瓶颈：SQLAlchemy 的异步模式依赖 asyncio 适配，复杂查询优化与连接池管理难以突破传统框架限制；Peewee 等轻量 ORM 专注同步场景，无法满足高并发下的异步协作需求；Django ORM 虽数据建模稳健，却因深度耦合框架、异步支持滞后，难以成为独立场景的最优解。

FastAPI 与 Tortoise-ORM 的组合正是为破解这些痛点而生：前者以高性能异步引擎和 OpenAPI 规范，适配云原生微服务架构；后者以 “轻量异步优先” 设计，支持多数据库方言并提供高效连接池管理。二者协同打通 “框架异步能力” 与 “ORM 执行效率” 的断层 ——FastAPI 的依赖注入与 Tortoise-ORM 的模型定义无缝衔接，异步路由与数据库操作形成流水线协作，让开发者在享受类型提示高效开发的同时，确保高并发下的服务稳定性。本文将聚焦这对 “异步拍档” 的实战实践，解析如何突破传统局限，实现开发效率与性能的双重提升。

二、FastAPI 与 Tortoise-ORM 的核心优势

在异步编程重塑后端开发范式的时代，FastAPI 与 Tortoise-ORM 分别以「API 开发效率」和「数据持久化优雅性」为支点，构建了一套完整的异步技术栈。两者的设计哲学既保持独立技术深度，又在工程实践中形成互补，成为现代Python后端开发的黄金组合。对于习惯于前后端分离的小伙伴来说应该会更适应。

2.1 FastAPI：异步优先的 API 开发利器

FastAPI 的诞生源于对 Python Web 开发效率与性能瓶颈的突破：随着异步编程在高并发、I/O 密集型场景的需求激增，传统框架如 Flask 的异步支持薄弱、Django 的同步机制低效等问题逐渐凸显，开发者亟需一种既能发挥 Python 类型提示优势、又能原生支持异步的现代框架。

2018 年，开发者 Sebastián Ramírez 结合自身经验，基于 Starlette 和 Pydantic 构建了 FastAPI，通过类型安全的设计、自动生成 API 文档、高效的异步引擎，解决了传统框架在开发效率与性能上的痛点。其高性能、易上手、强类型检查等特性迅速获得社区青睐，成为微服务、实时 API 等场景的首选框架，推动 Python Web 开发进入 “异步 + 类型安全” 的高效时代。

FastAPI 的设计哲学围绕 "高效开发" 与 "生产级健壮性" 展开，其技术优势可从四个维度解析：

✅异步原生支持：重构并发处理模型 

FastAPI 基于 Starlette 异步框架 构建，完全遵循 Python 异步编程规范（async/await），从底层架构支持非阻塞 I/O 操作。通过 uvloop（高性能事件循环）或原生 asyncio，将 I/O 操作与 CPU 计算分离，避免传统同步框架中线程上下文切换的开销。

这种设计使其在处理数据库查询、文件读写、第三方接口调用等 I/O 密集型任务时，能充分利用事件循环机制，实现单进程支撑 10k+ 并发连接 的高性能表现。 

因此一定程度上也算是python在web开发或者高并发环境开发的一次进步吧，毕竟长期以来python给我的印象就是工具语言或者机器学习类语言，并发开发场景下远不如Java，更不谈Go。

✅类型提示革命：构建编译级安全防线 

借助 Pydantic 数据模型，FastAPI 将 Python 的类型提示从「开发辅助」提升为「运行时校验」的核心能力：

from pydantic import BaseModel  
class UserCreate(BaseModel):  
    username: str  # 自动校验字符串长度（默认 0 < len <= 1e5）  
    age: int       # 自动拒绝非整数输入  

✅轻量架构：聚焦 API 开发的「精准打击」 

与 Django 等全功能框架相比，FastAPI 选择「做精不做全」的策略：

 2.2Tortoise-ORM：异步ORM新范式

Tortoise – ORM 的诞生源于 Python 异步 Web 开发浪潮下对高效数据库操作方案的渴求。传统 ORM 如 SQLAlchemy、Django ORM 多为同步设计，在高并发、I/O 密集场景下成为性能瓶颈。开发者构建高性能异步 Web 应用时，急需一款能与异步框架无缝集成、支持多数据库且 API 简洁的 ORM。

2018 年左右，社区开发者共同推出 Tortoise – ORM，专注异步操作。此后，它不断发展，增加数据库支持、优化 API 设计、提升性能与稳定性。其优势显著，异步特性适配异步框架，链式查询语法简洁直观，支持多数据库提供灵活选择，高效连接池管理保障资源利用。Tortoise – ORM 满足了开发者需求，成为 Python 异步开发中强大便捷的数据库操作解决方案。

作为首款原生支持异步操作的 Python ORM，Tortoise-ORM 重新定义了数据持久化的开发体验，核心优势体现在对传统 ORM 痛点的针对性解决，主要有如下几点：

✅非阻塞数据库操作 

传统 ORM（如 SQLAlchemy 同步模式、Django ORM）在执行数据库操作时会阻塞事件循环，导致异步框架性能优势无法发挥。而Tortoise-ORM 底层集成 asyncpg（PostgreSQL）与 aiomysql（MySQL）异步驱动，实现真正的 IO 与 CPU 并行处理：

# 异步创建用户（非阻塞）  
user = await User.create(username="test", email="test@example.com")  

目前Tortoise ORM 支持一下数据库的异步驱动使用:

 ✅极简语法设计

继承 Pydantic 的「声明式模型」设计哲学，Tortoise-ORM 的数据模型定义简洁直观，学习成本比 SQLAlchemy 异步模式降低 40%。在定义字段时Tortoise-ORM支持 CharField/IntField/DatetimeField 等常用类型，内置 auto_now_add/unique 等修饰符使用。

from tortoise.models import Model  
from tortoise import fields  

class User(Model):  
    id = fields.IntField(pk=True)  
    username = fields.CharField(max_length=50, unique=True)  
    create_time = fields.DatetimeField(auto_now_add=True)  # 自动填充创建时间  

除此以外，Tortoise-ORM也支持 filter()/order_by()/limit() 等链式操作，复杂查询可通过 Q 对象组合（类似 Django ORM） ：

# 查询最近 10 个活跃用户  
users = await User.filter(is_active=True).order_by("-create_time").limit(10)  

 ✅轻量级架构

与 SQLAlchemy（150k+ 行代码）相比，Tortoise-ORM 核心代码仅 10k 行级别，内存占用减少 30%，具有以下优势：

✅Tortoise-ORM与传统 ORM 对比 

 三、FastAPI 与 Tortoise-ORM在经典开发场景下的应用流程

这里我们结合最常见的教师、学生、课程、成绩等基础增删改查业务来实现整个项目，数据库采用pgsql，数据表如下：

/*
 Navicat Premium Dump SQL

 Source Server         : 本地PGSQL
 Source Server Type    : PostgreSQL
 Source Server Version : 170000 (170000)
 Source Host           : localhost:5432
 Source Catalog        : python_test
 Source Schema         : public

 Target Server Type    : PostgreSQL
 Target Server Version : 170000 (170000)
 File Encoding         : 65001

 Date: 10/04/2025 09:38:40
*/


-- ----------------------------
-- Table structure for teacher
-- ----------------------------
DROP TABLE IF EXISTS "public"."teacher";
CREATE TABLE "public"."teacher" (
  "tid" int4,
  "tname" text COLLATE "pg_catalog"."default"
)
;

-- ----------------------------
-- Records of teacher
-- ----------------------------
INSERT INTO "public"."teacher" VALUES (1, '张老师');
INSERT INTO "public"."teacher" VALUES (2, '李老师');
INSERT INTO "public"."teacher" VALUES (3, '张雪峰');
INSERT INTO "public"."teacher" VALUES (4, '刘广笔');
INSERT INTO "public"."teacher" VALUES (5, '赵六');
INSERT INTO "public"."teacher" VALUES (6, '李志月');

-- ----------------------------
-- Table structure for student
-- ----------------------------
DROP TABLE IF EXISTS "public"."student";
CREATE TABLE "public"."student" (
  "sid" int4,
  "sname" text COLLATE "pg_catalog"."default",
  "sage" int4,
  "ssex" text COLLATE "pg_catalog"."default"
)
;

-- ----------------------------
-- Records of student
-- ----------------------------
INSERT INTO "public"."student" VALUES (1, '张三', 22, '男');
INSERT INTO "public"."student" VALUES (2, '李四', 19, '男');
INSERT INTO "public"."student" VALUES (3, '王五', 20, '女');
INSERT INTO "public"."student" VALUES (4, '黎治跃', 21, '男');
INSERT INTO "public"."student" VALUES (5, '李帅', 21, '男');
INSERT INTO "public"."student" VALUES (6, '黄江小', 20, '女');
INSERT INTO "public"."student" VALUES (7, '菜菜懒', 19, '女');
INSERT INTO "public"."student" VALUES (8, '小花', 20, '女');
INSERT INTO "public"."student" VALUES (9, '闷油瓶', 25, '男');
INSERT INTO "public"."student" VALUES (10, '曾小贤', 28, '男');

-- ----------------------------
-- Table structure for sc
-- ----------------------------
DROP TABLE IF EXISTS "public"."sc";
CREATE TABLE "public"."sc" (
  "sid" int4,
  "cid" int4,
  "score" int4
)
;

-- ----------------------------
-- Records of sc
-- ----------------------------
INSERT INTO "public"."sc" VALUES (1, 2, 85);
INSERT INTO "public"."sc" VALUES (1, 5, 81);
INSERT INTO "public"."sc" VALUES (1, 6, 63);
INSERT INTO "public"."sc" VALUES (2, 5, 95);
INSERT INTO "public"."sc" VALUES (2, 6, 85);
INSERT INTO "public"."sc" VALUES (3, 1, 56);
INSERT INTO "public"."sc" VALUES (3, 5, 63);
INSERT INTO "public"."sc" VALUES (3, 6, 76);
INSERT INTO "public"."sc" VALUES (4, 1, 65);
INSERT INTO "public"."sc" VALUES (4, 5, 87);
INSERT INTO "public"."sc" VALUES (4, 8, 74);
INSERT INTO "public"."sc" VALUES (5, 1, 85);
INSERT INTO "public"."sc" VALUES (5, 5, 90);
INSERT INTO "public"."sc" VALUES (5, 9, 91);
INSERT INTO "public"."sc" VALUES (6, 1, 86);
INSERT INTO "public"."sc" VALUES (6, 5, 68);
INSERT INTO "public"."sc" VALUES (7, 1, 75);
INSERT INTO "public"."sc" VALUES (7, 5, 63);
INSERT INTO "public"."sc" VALUES (8, 1, 68);
INSERT INTO "public"."sc" VALUES (8, 5, 76);
INSERT INTO "public"."sc" VALUES (8, 8, 88);
INSERT INTO "public"."sc" VALUES (9, 1, 70);
INSERT INTO "public"."sc" VALUES (9, 5, 92);
INSERT INTO "public"."sc" VALUES (10, 1, 90);
INSERT INTO "public"."sc" VALUES (10, 5, 86);
INSERT INTO "public"."sc" VALUES (1, 1, 50);
INSERT INTO "public"."sc" VALUES (2, 1, 75);

通过python虚拟环境构建项目，这里构建步骤省略不计，但是需要注意提前装备好两个开发框架的依赖：

pip install fastapi uvicorn tortoise-orm pydantic psycopg2

准备好后即可开始项目搭建工作。 

Step1：模型定义

在 app/models/models.py 中定义Tortoise ORM模型，用于将 Python 类与数据库表进行映射。相当于java开发中的实体类，每个类对应一个数据库表，类的属性对应表的列。通过定义这些模型，我们可以使用 Python 代码来操作数据库，而不需要直接编写 SQL 语句。 

from tortoise.models import Model
from tortoise import fields


class Teacher(Model):
    """
    教师模型
    - tid: 教师表ID (主键)
    - tname: 教师姓名
    """
    tid = fields.IntField(pk=True)
    tname = fields.CharField(max_length=255, description="教师姓名")

    class Meta:
        table = "teacher"


class Student(Model):
    """
    学生模型
    - sid: 学生表ID (主键)
    - sname: 学生姓名
    - sage: 学生年龄
    - ssex: 学生性别
    """
    sid = fields.IntField(pk=True)
    sname = fields.CharField(max_length=255, description="学生姓名")
    sage = fields.IntField(description="学生年龄")
    ssex = fields.CharField(max_length=10, description="学生性别")

    class Meta:
        table = "student"


class SC(Model):
    """
    成绩模型
    - id: 成绩表ID (主键)
    - sid: 学生ID
    - cid: 课程ID
    - score: 分数
    """
    id = fields.IntField(pk=True)
    sid = fields.IntField(description="学生ID")
    cid = fields.IntField(description="课程ID")
    score = fields.IntField(description="分数")

    class Meta:
        table = "sc"


class Course(Model):
    """
    课程模型
    - cid: 课程表ID (主键)
    - cname: 课程名称
    - tid: 教师ID (外键)
    """
    cid = fields.IntField(pk=True)
    cname = fields.CharField(max_length=255, description="课程名称")
    tid = fields.IntField(description="教师ID")

    class Meta:
        table = "course"

 这里我们定义的就是对应的数据库中的四个数据表对象及字段信息。

Step2：定义Pydantic模型

在 app/schemas/schemas.py 中定义pydantic 模型，pydantic是一个用于数据解析和验证的 Python 库，特别适用于构建 API 和 Web 应用时处理请求和响应数据。

# schemas.py
from pydantic import BaseModel


class TeacherBase(BaseModel):
    tname: str


class TeacherCreate(TeacherBase):
    pass


class Teacher(TeacherBase):
    tid: int

    class Config:
        from_attributes = True  # 注意这里已经调整了


class StudentBase(BaseModel):
    sname: str
    sage: int
    ssex: str


class StudentCreate(StudentBase):
    pass


class Student(StudentBase):
    sid: int

    class Config:
        from_attributes = True


class SCBase(BaseModel):
    sid: int
    cid: int
    score: int


class SCCreate(SCBase):
    pass


class SC(SCBase):
    class Config:
        from_attributes = True


class CourseBase(BaseModel):
    cname: str
    tid: int


class CourseCreate(CourseBase):
    pass


class Course(CourseBase):
    cid: int

    class Config:
        from_attributes = True

这里定义了用于管理教师、学生和课程的信息以及他们之间的关系的数据模型。通过继承基本模型（如 TeacherBase、StudentBase、CourseBase），可以创建不同的子模型来适应不同的操作场景（如创建新记录）。设置 from_attributes = True 使得这些模型类可以直接从对象的属性初始化，便于与数据库交互，从而简化了数据处理流程。

Step3：项目配置

这里主要有四个方面，首先就是主要的数据库配置，由于我们使用的pgsql，而在异步项目中我们可以使用psql的异步驱动来配置：

# config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # 数据库连接URL，从.env文件中读取
    DB_USER: str = "postgres"
    DB_PASSWORD: str = "123456"
    DB_HOST: str = "localhost"
    DB_PORT: int = 5432
    DB_NAME: str = "python_test"
    # 日志级别，默认值为INFO
    log_level: str = "INFO"

    class Config:
        # 指定.env文件路径
        env_file = ".env"
        # 允许使用环境变量前缀
        env_file_encoding = 'utf-8'


# 创建配置实例
settings = Settings()

 这里我将数据库配置分开进行处理，config.py记录数据库配置信息，用的是明文信息，实际开发中我们可以通过环境配置文件读取，然后配置数据库初始化信息database.py：

from tortoise import Tortoise
from tortoise.contrib.fastapi import register_tortoise
from app.core.config import settings
from fastapi import FastAPI
import logging

TORTOISE_ORM = {
    "connections": {
        "default": f"asyncpg://{settings.DB_USER}:{settings.DB_PASSWORD}@{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}"
    },
    "apps": {
        "models": {
            "models": ["app.models.models"],
            "default_connection": "default",
        },
    },
    "use_tz": True,
    "timezone": "Asia/Shanghai"
}


async def init_database():
    """
    初始化数据库连接
    """
    # 设置 Tortoise ORM 的日志级别为 DEBUG
    logger = logging.getLogger("tortoise")
    logger.setLevel(logging.DEBUG)
    
    await Tortoise.init(
        db_url=TORTOISE_ORM["connections"]["default"],
        modules={"models": TORTOISE_ORM["apps"]["models"]["models"]},
        timezone=TORTOISE_ORM["timezone"],
        use_tz=TORTOISE_ORM["use_tz"]
    )
    # 自动根据models定义的model生成数据库表结构
    # await Tortoise.generate_schemas()


async def close_database():
    """
    关闭数据库连接
    """
    await Tortoise.close_connections()


def register_database(app: FastAPI):
    """
    注册数据库到FastAPI应用
    
    Args:
        app: FastAPI应用实例
    """
    register_tortoise(
        app,
        config=TORTOISE_ORM,
        generate_schemas=False,
        add_exception_handlers=True,
    )

需要使用时注册到项目中即可进行使用。除此以外，日志记录在日常开发中也是不可或缺的一部分，这里我们为了检查业务执行情况，将运行日志、sql执行日志与Tortoise ORM日志分为3个单独文件保存到本地：

import logging
import os
from logging.handlers import RotatingFileHandler, WatchedFileHandler
from pathlib import Path
from app.core.config import settings

# 创建日志目录
log_dir = Path("logs")
log_dir.mkdir(exist_ok=True)

# 定义日志格式
log_format = logging.Formatter(
    "%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s"
)

# SQL日志格式
sql_log_format = logging.Formatter(
    "%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s"
)


def setup_logger(name, log_file, level=logging.DEBUG, formatter=None):
    """
    设置并返回一个自定义的日志记录器
    
    Args:
        name: 日志记录器名称
        log_file: 日志文件路径
        level: 日志级别，默认为DEBUG
        formatter: 自定义的日志格式，如果为None则使用默认格式
    
    Returns:
        logging.Logger: 配置好的日志记录器
    """
    logger = logging.getLogger(name)
    logger.setLevel(level)

    # 如果已经有处理器，不重复添加
    if logger.handlers:
        return logger

    # 创建文件处理器
    file_handler = RotatingFileHandler(
        log_file,
        maxBytes=10 * 1024 * 1024,  # 10MB
        backupCount=7,
        encoding='utf-8'
    )
    file_handler.setFormatter(formatter or log_format)

    # 创建控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter or log_format)

    # 添加处理器到日志记录器
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)

    return logger


# 创建应用程序日志记录器
app_logger = setup_logger(
    "app",
    log_dir / "app.log",
    level=getattr(logging, settings.log_level.upper())
)

# 创建Tortoise ORM日志记录器
tortoise_logger = setup_logger(
    "tortoise",
    log_dir / "tortoise.log",
    level=getattr(logging, settings.log_level.upper())
)

# 创建数据库SQL日志记录器
db_logger = logging.getLogger("db_client")
db_logger.setLevel(logging.DEBUG)  # 设置为DEBUG级别以记录所有SQL查询

# 创建SQL日志文件处理器
sql_handler = WatchedFileHandler(
    filename=log_dir / "db_client.log",
    encoding='utf-8'
)
sql_handler.setFormatter(sql_log_format)
db_logger.addHandler(sql_handler)

# 确保SQL日志记录器不会向上传播到根记录器
db_logger.propagate = False

为了统一规范接口响应数据和异常，我们可以像Java开发中在这里进行定义，首先设置统一返回值类result.py：

from typing import Generic, TypeVar, Optional
from pydantic import BaseModel

T = TypeVar('T')


class Result(BaseModel, Generic[T]):
    """
    统一响应模型基类
    """
    code: int
    message: str
    data: Optional[T] = None


class SuccessResult(Result[T]):
    """
    成功响应模型
    """

    def __init__(self, data: T = None, message: str = "操作成功"):
        super().__init__(code=200, message=message, data=data)


class ErrorResult(Result[T]):
    """
    错误响应模型
    """

    def __init__(self, code: int = 400, message: str = "操作失败", data: T = None):
        super().__init__(code=code, message=message, data=data)


def success(data: T = None, message: str = "操作成功") -> SuccessResult[T]:
    """
    快速创建成功响应
    """
    return SuccessResult(data=data, message=message)


def error(code: int = 400, message: str = "操作失败", data: T = None) -> ErrorResult[T]:
    """
    快速创建错误响应
    """
    return ErrorResult(code=code, message=message, data=data)

这里我们主要使用了Python的类型提示（type hints）和Pydantic库来定义一个通用的响应模型，并在此基础上区分成功和错误的响应，让我们接口响应数据格式更规范。然后配置一些可能在项目运行中出现的全局异常处理exception.py：

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from app.core.result import error
import logging
from typing import Any

logger = logging.getLogger("app")


class AppException(Exception):
    """基础应用异常"""

    def __init__(self, code: int = 400, message: str = "请求错误", detail: Any = None):
        self.code = code
        self.message = message
        self.detail = detail


class NotFoundException(AppException):
    """资源未找到异常"""

    def __init__(self, message: str = "资源未找到", detail: Any = None):
        super().__init__(404, message, detail)



class ValidationException(AppException):
    """参数验证异常"""

    def __init__(self, message: str = "参数验证失败", detail: Any = None):
        super().__init__(422, message, detail)


async def global_exception_handler(request: Request, exc: Exception):
    """全局异常处理器"""
    logger.error(f"请求异常: {str(exc)}", exc_info=True)

    if isinstance(exc, AppException):
        return JSONResponse(
            status_code=exc.code,
            content=error(code=exc.code, message=exc.message, data=exc.detail).dict()
        )
    elif isinstance(exc, HTTPException):
        return JSONResponse(
            status_code=exc.status_code,
            content=error(code=exc.status_code, message=str(exc.detail)).dict()
        )

    # 处理未捕获的异常
    return JSONResponse(
        status_code=500,
        content=error(code=500, message="服务器内部错误").dict()
    )


def register_exception_handlers(app):
    """注册全局异常处理器"""
    app.add_exception_handler(Exception, global_exception_handler)

Step4：定义具体的业务逻辑

这里就相当于MVC开发中的service层，但需要使用异步函数。用于处理与教师、学生、课程和成绩相关的CRUD（创建、读取、更新、删除）操作。每个函数都接受特定类型的参数并返回一个字典，该字典包含操作的结果信息。

from typing import Dict, Any

from app.core.result import success, error
from app.models.models import Teacher, Student, SC, Course
from app.schemas.schemas import TeacherCreate, StudentCreate, SCCreate, CourseCreate
from fastapi.encoders import jsonable_encoder  #  解决序列化问题


async def create_teacher(teacher: TeacherCreate) -> Dict[str, Any]:
    """
    创建教师业务逻辑
    :param teacher: 教师创建信息
    :return: 创建的教师对象
    """
    try:
        teacher_obj = await Teacher.create(**teacher.dict())
        return success(jsonable_encoder(teacher_obj))
    except Exception as e:
        return error(400, f"创建教师失败: {str(e)}")


async def get_teacher(teacher_id: int) -> Dict[str, Any]:
    """
    获取教师信息业务逻辑
    :param teacher_id: 教师ID
    :return: 教师对象
    """
    try:
        teacher = await Teacher.get(tid=teacher_id)
        if not teacher:
            return error(404, "教师未找到", {"teacher_id": teacher_id})
        return success(jsonable_encoder(teacher))
    except Exception as e:
        return error(400, f"获取教师信息失败: {str(e)}")


async def update_teacher(teacher_id: int, teacher: TeacherCreate) -> Dict[str, Any]:
    """
    更新教师信息业务逻辑
    :param teacher_id: 教师ID
    :param teacher: 教师更新信息
    :return: 更新后的教师对象
    """
    try:
        db_teacher = await Teacher.get(tid=teacher_id)
        if not db_teacher:
            return error(404, "教师未找到", {"teacher_id": teacher_id})
        updated_teacher = await db_teacher.update_from_dict(teacher.dict()).save()
        return success(jsonable_encoder(updated_teacher))
    except Exception as e:
        return error(400, f"更新教师信息失败: {str(e)}")


async def delete_teacher(teacher_id: int) -> Dict[str, str]:
    """
    删除教师业务逻辑
    :param teacher_id: 教师ID
    :return: 删除结果
    """
    try:
        db_teacher = await Teacher.get(tid=teacher_id)
        if not db_teacher:
            return error(404, "教师未找到", {"teacher_id": teacher_id})
        await db_teacher.delete()
        return success(message="教师删除成功")
    except Exception as e:
        return error(400, f"删除教师失败: {str(e)}")


async def create_student(student: StudentCreate) -> Dict[str, Any]:
    """
    创建学生业务逻辑
    :param student: 学生创建信息
    :return: 创建的学生对象
    """
    try:
        student_obj = await Student.create(**student.dict())
        return success(jsonable_encoder(student_obj))
    except Exception as e:
        return error(400, f"创建学生失败: {str(e)}")


async def get_student(student_id: int) -> Dict[str, Any]:
    """
    获取学生信息业务逻辑
    :param student_id: 学生ID
    :return: 学生对象
    """
    try:
        student = await Student.get(sid=student_id)
        if not student:
            return error(404, "学生未找到", {"student_id": student_id})
        return success(jsonable_encoder(student))
    except Exception as e:
        return error(400, f"获取学生信息失败: {str(e)}")


async def update_student(student_id: int, student: StudentCreate) -> Dict[str, Any]:
    """
    更新学生信息业务逻辑
    :param student_id: 学生ID
    :param student: 学生更新信息
    :return: 更新后的学生对象
    """
    try:
        db_student = await Student.get(sid=student_id)
        if not db_student:
            return error(404, "学生未找到", {"student_id": student_id})
        updated_student = await db_student.update_from_dict(student.dict()).save()
        return success(jsonable_encoder(updated_student))
    except Exception as e:
        return error(400, f"更新学生信息失败: {str(e)}")


async def delete_student(student_id: int) -> Dict[str, str]:
    """
    删除学生业务逻辑
    :param student_id: 学生ID
    :return: 删除结果
    """
    try:
        db_student = await Student.get(sid=student_id)
        if not db_student:
            return error(404, "学生未找到", {"student_id": student_id})
        await db_student.delete()
        return success(message="学生删除成功")
    except Exception as e:
        return error(400, f"删除学生失败: {str(e)}")


async def create_course(course: CourseCreate) -> Dict[str, Any]:
    """
    创建课程业务逻辑
    :param course: 课程创建信息
    :return: 创建的课程对象
    """
    try:
        course_obj = await Course.create(**course.dict())
        return success(jsonable_encoder(course_obj))
    except Exception as e:
        return error(400, f"创建课程失败: {str(e)}")


async def get_course(course_id: int) -> Dict[str, Any]:
    """
    获取课程信息业务逻辑
    :param course_id: 课程ID
    :return: 课程对象
    """
    try:
        course = await Course.get(cid=course_id)
        if not course:
            return error(404, "课程未找到", {"course_id": course_id})
        return success(jsonable_encoder(course))
    except Exception as e:
        return error(400, f"获取课程信息失败: {str(e)}")


async def create_score(score: SCCreate) -> Dict[str, Any]:
    """
    创建成绩业务逻辑
    :param score: 成绩创建信息
    :return: 创建的成绩对象
    """
    try:
        score_obj = await SC.create(**score.dict())
        return success(jsonable_encoder(score_obj))
    except Exception as e:
        return error(400, f"创建成绩失败: {str(e)}")


async def update_course(course_id: int, course: CourseCreate) -> Dict[str, Any]:
    """
    更新课程信息业务逻辑
    :param course_id: 课程ID
    :param course: 课程更新信息
    :return: 更新后的课程对象
    """
    try:
        db_course = await Course.get(cid=course_id)
        if not db_course:
            return error(404, "课程未找到", {"course_id": course_id})
        updated_course = await db_course.update_from_dict(course.dict()).save()
        return success(jsonable_encoder(updated_course))
    except Exception as e:
        return error(400, f"更新课程信息失败: {str(e)}")


async def delete_course(course_id: int) -> Dict[str, str]:
    """
    删除课程业务逻辑
    :param course_id: 课程ID
    :return: 删除结果
    """
    try:
        db_course = await Course.get(cid=course_id)
        if not db_course:
            return error(404, "课程未找到", {"course_id": course_id})
        await db_course.delete()
        return success(message="课程删除成功")
    except Exception as e:
        return error(400, f"删除课程失败: {str(e)}")


async def get_courses_list() -> Dict[str, Any]:
    """
    获取所有课程列表业务逻辑
    :return: 所有课程列表
    """
    try:
        courses = await Course.all()
        if not courses:
            return error(404, "课程信息获取失败")
        return success(jsonable_encoder(courses))
    except Exception as e:
        return error(400, f"获取课程列表失败: {str(e)}")


async def get_scores(student_id: int) -> Dict[str, Any]:
    """
    获取学生成绩业务逻辑
    :param student_id: 学生ID
    :return: 该学生的所有成绩列表
    """
    try:
        scores = await SC.filter(sid=student_id).all()
        if not scores:
            return error(404, "未找到该学生的成绩记录", {"student_id": student_id})
        return success(jsonable_encoder(scores))
    except Exception as e:
        return error(400, f"获取成绩信息失败: {str(e)}")


async def get_student_with_scores(student_id: int) -> Dict[str, Any]:
    """
    获取学生成绩详细信息，连表查询
    :param student_id: 学生ID
    :return: 学生及其成绩列表
    """
    try:
        # 获取学生基本信息
        student = await Student.get(sid=student_id)
        if not student:
            return error(404, "学生未找到", {"student_id": student_id})

        # 获取学生所有成绩记录
        scores = await SC.filter(sid=student_id).all()
        if not scores:
            return error(404, "该学生暂无成绩记录", {"student_id": student_id})

        # 获取每门课程的信息
        course_scores = []
        for score in scores:
            course = await Course.get(cid=score.cid)
            if course:
                course_scores.append({
                    'course_id': course.cid,
                    'course_name': course.cname,
                    'score': score.score,
                })

        result = {
            'student_id': student.sid,
            'student_name': student.sname,
            'student_age': student.sage,
            'student_sex': student.ssex,
            'courses': course_scores,
            'average_score': sum(course['score'] for course in course_scores) / len(
                course_scores) if course_scores else 0
        }
        return success(jsonable_encoder(result))
    except Exception as e:
        return error(400, f"获取学生成绩信息失败: {str(e)}")


async def get_teacher_with_courses(teacher_id: int) -> Dict[str, Any]:
    """
    获取教师所教课程信息，连表查询
    :param teacher_id: 教师ID
    :return: 教师及其所教课程列表
    """
    try:
        # 获取教师基本信息
        teacher = await Teacher.get(tid=teacher_id)
        if not teacher:
            return error(404, "教师未找到", {"teacher_id": teacher_id})

        # 获取教师所教的所有课程
        courses = await Course.filter(tid=teacher_id).all()
        if not courses:
            return error(404, "该教师暂无授课课程", {"teacher_id": teacher_id})

        # 获取每门课程的选课学生数量
        course_details = []
        for course in courses:
            student_count = await SC.filter(cid=course.cid).count()
            course_details.append({
                'course_id': course.cid,
                'course_name': course.cname,
                'student_count': student_count
            })

        result = {
            'teacher_id': teacher.tid,
            'teacher_name': teacher.tname,
            'courses': course_details,
            'total_courses': len(course_details),
            'total_students': sum(course['student_count'] for course in course_details)
        }
        return success(jsonable_encoder(result))
    except Exception as e:
        return error(400, f"获取教师课程信息失败: {str(e)}")


async def get_course_students(course_id: int) -> Dict[str, Any]:
    """
    获取课程的学生列表信息
    :param course_id: 课程ID
    :return: 课程及其学生列表
    """
    try:
        # 获取课程基本信息
        course = await Course.get(cid=course_id)
        if not course:
            return error(404, "课程未找到", {"course_id": course_id})

        # 获取选修该课程的所有学生成绩记录
        scores = await SC.filter(cid=course_id).all()
        if not scores:
            return error(404, "暂无学生选修该课程", {"course_id": course_id})

        # 获取每个学生的详细信息
        student_details = []
        for score in scores:
            student = await Student.get(sid=score.sid)
            if student:
                student_details.append({
                    'student_id': student.sid,
                    'student_name': student.sname,
                    'student_age': student.sage,
                    'student_sex': student.ssex,
                    'score': score.score
                })

        # 计算课程统计信息
        total_students = len(student_details)
        average_score = sum(
            student['score'] for student in student_details) / total_students if total_students > 0 else 0
        max_score = max(student['score'] for student in student_details) if student_details else 0
        min_score = min(student['score'] for student in student_details) if student_details else 0

        result = {
            'course_id': course.cid,
            'course_name': course.cname,
            'teacher_id': course.tid,
            'students': student_details,
            'statistics': {
                'total_students': total_students,
                'average_score': average_score,
                'max_score': max_score,
                'min_score': min_score
            }
        }
        return success(jsonable_encoder(result))
    except Exception as e:
        return error(400, f"获取课程学生列表失败: {str(e)}")

 Step5：创建后端接口（路由）

根据引用业务实现具体的python后端接口，可类比于springboot中的controller，但是在fastapi中它也叫路由。

from fastapi import APIRouter

# 在导入部分添加课程相关服务
from app.application.services import (
    create_teacher, get_teacher, update_teacher, delete_teacher,
    create_student, get_student, update_student, delete_student,
    create_score, get_scores,
    get_student_with_scores,
    get_teacher_with_courses, create_course, get_course, delete_course,
    get_course_students, update_course, get_courses_list
)
from app.core.result import Result
from app.schemas.schemas import TeacherCreate, StudentCreate, SCCreate, CourseCreate

router = APIRouter(prefix="/api/v1", tags=["API接口"])


@router.post("/teachers/", response_model=Result, summary="创建教师", description="创建一个新的教师记录")
async def create_teacher_endpoint(teacher: TeacherCreate):
    return await create_teacher(teacher)


@router.get("/teachers/{teacher_id}", response_model=Result, summary="获取教师信息",
            description="根据教师ID获取教师信息")
async def get_teacher_endpoint(teacher_id: int):
    return await get_teacher(teacher_id)


@router.post("/students/add", response_model=Result, summary="创建学生", description="创建一个新的学生记录")
async def create_student_endpoint(student: StudentCreate):
    return await create_student(student)


@router.get("/students/{student_id}", response_model=Result, summary="获取学生信息",
            description="根据学生ID获取学生信息")
async def get_student_endpoint(student_id: int):
    return await get_student(student_id)


@router.post("/scores/", response_model=Result, summary="创建成绩", description="创建一个新的成绩记录")
async def create_score_endpoint(score: SCCreate):
    return await create_score(score)


@router.get("/scores/{student_id}", response_model=Result, summary="获取学生成绩",
            description="根据学生ID获取该学生的所有成绩")
async def get_scores_endpoint(student_id: int):
    return await get_scores(student_id)


@router.put("/teachers/{teacher_id}", response_model=Result, summary="更新教师信息")
async def update_teacher_endpoint(teacher_id: int, teacher: TeacherCreate):
    return await update_teacher(teacher_id, teacher)


@router.delete("/teachers/{teacher_id}", response_model=Result, summary="删除教师")
async def delete_teacher_endpoint(teacher_id: int):
    return await delete_teacher(teacher_id)


@router.put("/students/update/{student_id}", response_model=Result, summary="更新学生信息")
async def update_student_endpoint(student_id: int, student: StudentCreate):
    return await update_student(student_id, student)


@router.delete("/students/{student_id}", response_model=Result, summary="删除学生")
async def delete_student_endpoint(student_id: int):
    return await delete_student(student_id)


@router.get("/students/{student_id}/scores", response_model=Result, summary="获取学生及其所有成绩",
            description="根据学生ID获取学生信息和所有成绩记录")
async def get_student_with_scores_endpoint(student_id: int):
    return await get_student_with_scores(student_id)


@router.get("/teachers/{teacher_id}/courses", response_model=Result, summary="获取教师及其授课课程",
            description="根据教师ID获取教师信息和所授课程")
async def get_teacher_with_courses_endpoint(teacher_id: int):
    return await get_teacher_with_courses(teacher_id)


@router.post("/courses/", response_model=Result, summary="创建课程", description="创建一个新的课程记录")
async def create_course_endpoint(course: CourseCreate):
    return await create_course(course)


@router.get("/courses/{course_id}", response_model=Result, summary="获取课程信息")
async def get_course_endpoint(course_id: int):
    return await get_course(course_id)


@router.put("/courses/update/{course_id}", response_model=Result, summary="更新课程信息")
async def update_course_endpoint(course_id: int, course: CourseCreate):
    return await update_course(course_id, course)


@router.delete("/courses/del/{course_id}", response_model=Result, summary="删除课程")
async def delete_course_endpoint(course_id: int):
    return await delete_course(course_id)


@router.delete("/courses/list", response_model=Result, summary="获取所有课程")
async def course_list():
    return await get_courses_list()


@router.get("/courses/{course_id}/students", response_model=Result, summary="获取课程及其学生列表",
            description="根据课程ID获取课程信息和选修该课程的学生列表")
async def get_course_students_endpoint(course_id: int):
    return await get_course_students(course_id)

 Step6：项目运行及测试

在main.py中配置之前咱们配置过的数据库初始信息，全局异常等信息，有需求的可以加上swagger的一些信息，但是由于cdn原因，fastapi中swagger访问会出现点问题，需要下载相应的静态文件到本地，这里不做示例，大家可以根据自己情况自行配置：

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.core.logger import app_logger, tortoise_logger, db_logger
from app.infrastructure.database import init_database, close_database, register_database
from app.routers import api


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行
    app_logger.info("Application startup")
    tortoise_logger.info("Tortoise ORM startup")
    await init_database()
    yield
    # 关闭时执行
    app_logger.info("Application shutdown")
    tortoise_logger.info("Tortoise ORM shutdown")
    await close_database()


app = FastAPI(
    title="Student Management System",
    version="1.0.0",
    lifespan=lifespan
)

# 注册路由
app.include_router(api.router)

# 注册数据库
register_database(app)

# 配置 Tortoise ORM 日志
import logging
from tortoise.log import logger

# 设置 Tortoise ORM 的日志级别
logger.setLevel(logging.DEBUG)  # 设置为DEBUG级别以记录所有SQL查询

# 添加自定义处理器到 Tortoise ORM 的日志记录器
for handler in db_logger.handlers:
    logger.addHandler(handler)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8009)

然后通过测试工具对fastapi的这些接口测试：

完成后我们可以通过控制台日志或者日志文件查看到orm框架在执行业务时的sql信息了，这样方便我们出现问题可以方便检查：

INFO:     Started server process [2712]
INFO:     Waiting for application startup.
2025-04-12 20:21:03,974 - tortoise - DEBUG - [__init__.py:505] - Tortoise-ORM startup
    connections: {'default': 'asyncpg://postgres:12***@localhost:5432/python_test'}
    apps: {'models': {'models': ['app.models.models'], 'default_connection': 'default'}}
2025-04-12 20:21:04,077 - tortoise - INFO - [__init__.py:163] - Tortoise-ORM started, {'default': <tortoise.backends.asyncpg.client.AsyncpgDBClient object at 0x000001F73A3DCE90>}, {'models': {'Course': <class 'app.models.models.Course'>, 'SC': <class 'app.models.models.SC'>, 'Student': <class 'app.models.models.Student'>, 'Teacher': <class 'app.models.models.Teacher'>}}
2025-04-12 20:21:04,077 - app - INFO - [main.py:13] - Application startup
2025-04-12 20:21:04,078 - tortoise - INFO - [main.py:14] - Tortoise ORM startup
2025-04-12 20:21:04,078 - tortoise - DEBUG - [__init__.py:505] - Tortoise-ORM startup
    connections: {'default': {'engine': 'tortoise.backends.asyncpg', 'credentials': {'port': 5432, 'database': 'python_test', 'host': 'localhost', 'user': 'postgres', 'password': '12***'}}}
    apps: {'models': {'models': ['app.models.models'], 'default_connection': 'default'}}
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8009 (Press CTRL+C to quit)
2025-04-12 20:21:11,622 - tortoise.db_client - DEBUG - [client.py:62] - Created connection pool <asyncpg.pool.Pool object at 0x000001F73A517100> with params: {'host': 'localhost', 'port': 5432, 'user': 'postgres', 'database': 'python_test', 'min_size': 1, 'max_size': 5, 'connection_class': <class 'asyncpg.connection.Connection'>, 'loop': None, 'server_settings': {}}
2025-04-12 20:21:11,623 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "tid","tname" FROM "teacher" WHERE "tid"=$1 LIMIT $2: [5, 2]
INFO:     127.0.0.1:64560 - "GET /api/v1/teachers/5 HTTP/1.1" 200 OK
2025-04-12 20:21:12,052 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "sage","sname","ssex","sid" FROM "student" WHERE "sid"=$1 LIMIT $2: [3, 2]
INFO:     127.0.0.1:64560 - "GET /api/v1/students/3 HTTP/1.1" 200 OK
2025-04-12 20:21:12,475 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "cid","id","score","sid" FROM "sc" WHERE "sid"=$1: [3]
INFO:     127.0.0.1:64560 - "GET /api/v1/scores/3 HTTP/1.1" 200 OK
2025-04-12 20:21:12,873 - tortoise.db_client - DEBUG - [client.py:110] - INSERT INTO "student" ("sname","sage","ssex") VALUES ($1,$2,$3) RETURNING "sid": ['赵东来', 22, '男']
INFO:     127.0.0.1:64560 - "POST /api/v1/students/add HTTP/1.1" 200 OK
2025-04-12 20:21:13,288 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "sage","sname","ssex","sid" FROM "student" WHERE "sid"=$1 LIMIT $2: [3, 2]
2025-04-12 20:21:13,289 - tortoise.db_client - DEBUG - [client.py:132] - UPDATE "student" SET "sname"=$1,"sage"=$2,"ssex"=$3 WHERE "sid"=$4: ['黎治跃', 25, '男', 3]
INFO:     127.0.0.1:64560 - "PUT /api/v1/students/update/3 HTTP/1.1" 200 OK
2025-04-12 20:21:13,711 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "cname","cid","tid" FROM "course" WHERE "cid"=$1 LIMIT $2: [2, 2]
2025-04-12 20:21:13,713 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "cid","id","score","sid" FROM "sc" WHERE "cid"=$1: [2]
2025-04-12 20:21:13,714 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "sage","sname","ssex","sid" FROM "student" WHERE "sid"=$1 LIMIT $2: [1, 2]
2025-04-12 20:21:13,715 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "sage","sname","ssex","sid" FROM "student" WHERE "sid"=$1 LIMIT $2: [5, 2]
2025-04-12 20:21:13,716 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "sage","sname","ssex","sid" FROM "student" WHERE "sid"=$1 LIMIT $2: [1, 2]
INFO:     127.0.0.1:64560 - "GET /api/v1/courses/2/students HTTP/1.1" 200 OK
2025-04-12 20:21:14,154 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "tid","tname" FROM "teacher" WHERE "tid"=$1 LIMIT $2: [4, 2]
2025-04-12 20:21:14,155 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "cname","cid","tid" FROM "course" WHERE "tid"=$1: [4]
2025-04-12 20:21:14,156 - tortoise.db_client - DEBUG - [client.py:132] - SELECT COUNT(*) FROM "sc" WHERE "cid"=$1: [6]
2025-04-12 20:21:14,158 - tortoise.db_client - DEBUG - [client.py:132] - SELECT COUNT(*) FROM "sc" WHERE "cid"=$1: [10]
2025-04-12 20:21:14,159 - tortoise.db_client - DEBUG - [client.py:132] - SELECT COUNT(*) FROM "sc" WHERE "cid"=$1: [11]
INFO:     127.0.0.1:64560 - "GET /api/v1/teachers/4/courses HTTP/1.1" 200 OK
2025-04-12 20:21:14,583 - tortoise.db_client - DEBUG - [client.py:110] - INSERT INTO "course" ("cname","tid") VALUES ($1,$2) RETURNING "cid": ['毛泽东选集', 3]
INFO:     127.0.0.1:64560 - "POST /api/v1/courses/ HTTP/1.1" 200 OK
2025-04-12 20:21:14,976 - tortoise.db_client - DEBUG - [client.py:132] - SELECT "cname","cid","tid" FROM "course" WHERE "cid"=$1 LIMIT $2: [31, 2]
2025-04-12 20:21:14,977 - tortoise.db_client - DEBUG - [client.py:132] - DELETE FROM "course" WHERE "cid"=$1: [31]
INFO:     127.0.0.1:64560 - "DELETE /api/v1/courses/del/31 HTTP/1.1" 200 OK

并且完成了响应数据的规范化：

四、总结 

 可以看到，其实FastAPI 与 Tortoise-ORM 的组合在 Python Web 开发领域还是比较方便的，特别是对于纯后端开发者来说，接受起来并不太困难。

虽然本文并未对其在并发开发中的效能进行具体介绍，但是FastAPI 具备高效处理高并发请求的能力，Tortoise-ORM 则在异步操作方面性能卓越，二者协同构建了完整的异步链路，相信可以有效提升了系统的吞吐量，降低了响应延迟。当然了肯定不能和Go语言相对比，具体测试大家可以看一下知名性能评测博主 Anton Putra Python (FastAPI) vs Go (Golang) Performance Benchmark。

但是我认为也算是Python Web开发的一次提升吧。

此外呢，该组合在开发效率、代码维护及性能优化等方面表现突出，尤其适用于微服务、物联网（IoT）、实时系统等高要求场景，为 Python Web 开发者提供了全面优化的开发体验，无疑是当前构建高性能应用的理想选择。感兴趣的小伙伴可以试试~

作者：深情不及里子