FastAPI 简介

FastAPI 是一个现代、快速、类型友好的 Python Web 框架，适合做接口服务。它基于 ASGI 标准，天然支持异步编程。

什么是 async/await？

在 Python 中，async 和 await 是处理异步操作的关键字：

• async def：定义一个异步函数
• await：等待异步任务完成，但不会阻塞整个线程

# 同步写法（传统方式）
def list_books():
    return service.list_books()

# 异步写法（适用于异步数据库或外部请求）
async def list_books():
    result = await service.list_books_async()
    return result

tips：只有当内部调用的是异步库（如 asyncpg、httpx 等）时，async/await 才能带来真正的性能收益。如果你的代码里都是同步操作（如普通的 SQLAlchemy），使用 async 反而会增加开销。

---

如何启动一个 FastAPI 项目

创建应用实例

app = FastAPI() 是 FastAPI 应用的入口，所有路由、依赖、中间件都注册在它上面：

from fastapi import FastAPI

app = FastAPI()

启动项目

使用 Uvicorn 运行应用

# 开发环境（带热重载）
uvicorn main:app --reload

# 生产环境
uvicorn main:app --host 0.0.0.0 --port 8000

Uvicorn 的作用：它是 ASGI 服务器，负责接收 HTTP 请求并转发给 FastAPI 应用处理，同时支持异步并发。简单理解一下就是Uvicorn是用来启动FastApi项目的服务器。

写一个 GET 请求

from fastapi import FastAPI

app = FastAPI()

@app.get("/ping")
def ping():
    return {"message": "pong"}

启动后访问 http://localhost:8000/ping，你会看到 {"message": "pong"}。

---

什么是 Python 类型注解

FastAPI 的强大之处很大程度上依赖于 Python 的类型注解。

什么是类型注解？

类型注解是 Python 3.5+ 引入的特性，用于标注变量、函数参数和返回值的类型：

# 没有类型注解（传统写法）
def add(a, b):
    return a + b

# 有类型注解（推荐写法）
def add(a: int, b: int) -> int:
    return a + b

基础类型注解

from typing import Optional, List, Dict, Union

# 基本类型
name: str = "FastAPI"
count: int = 100
price: float = 99.9
is_valid: bool = True

# Optional：可选类型（可以是 None）
description: Optional[str] = None

# List：列表类型
tags: List[str] = ["python", "fastapi", "web"]

# Dict：字典类型
metadata: Dict[str, str] = {"version": "1.0", "author": "tom"}

# Union：联合类型（可以是多种类型之一）
value: Union[int, str] = 100  # 可以是 int 或 str

写一个简单的Demo

这个Demo采用了分层架构：Router → Service → Repository → Model。

项目结构

FastApiDemo/
├── main.py              # 应用入口
├── database.py          # 数据库配置
├── models/
│   └── book.py          # 数据模型（SQLAlchemy）
├── schemas/
│   └── book_schema.py   # Pydantic 模型（数据校验）
├── repositories/
│   └── book_repository.py  # 数据访问层
├── services/
│   └── book_service.py  # 业务逻辑层
└── routers/
    └── book_router.py   # 路由层（API 接口）

1. 数据库配置（database.py）

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

DATABASE_URL = "sqlite:///database.db"

# 创建数据库引擎
engine = create_engine(DATABASE_URL)
# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 依赖注入函数：每个请求创建一个数据库会话
def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db
        db.commit()
    finally:
        db.close()

2. 数据模型（models/book.py）

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Book(Base):
    __tablename__ = 'book'
    
    id: int = Column(Integer, primary_key=True)
    name: str = Column(String, nullable=False, index=True)
    author: str = Column(String, nullable=False)

3. Pydantic 模型（schemas/book_schema.py）

from pydantic import BaseModel
from typing import Optional

# 创建图书时的请求模型
class BookCreate(BaseModel):
    title: str
    author: str

# 返回给客户端的响应模型
class BookResponse(BookCreate):
    id: int

    class Config:
        orm_mode = True  # 允许从 SQLAlchemy 模型直接转换

4. 数据访问层（repositories/book_repository.py）

from typing import Optional, List
from sqlalchemy.orm import Session
from models.book import Book

class BookRepository:

    @staticmethod
    def create(db: Session, book_data) -> Book:
        book = Book(**book_data.dict())
        db.add(book)
        db.commit()
        db.refresh(book)
        return book

    @staticmethod
    def get(db: Session, book_id: int) -> Optional[Book]:
        return db.query(Book).filter(Book.id == book_id).first()

    @staticmethod
    def get_all(db: Session, skip: int = 0, limit: int = 100) -> List[Book]:
        return db.query(Book).offset(skip).limit(limit).all()

    @staticmethod
    def update_book_name(db: Session, book_id: int, book_data) -> Optional[Book]:
        book = db.query(Book).filter(Book.id == book_id).first()
        if book:
            book.name = book_data.name
            db.commit()
            db.refresh(book)
        return book

    @staticmethod
    def delete_book(db: Session, book_id: int) -> Optional[Book]:
        book = db.query(Book).filter(Book.id == book_id).first()
        if book:
            db.delete(book)
            db.commit()
        return book

5. 业务逻辑层（services/book_service.py）

from typing import List
from sqlalchemy.orm import Session
from models.book import Book
from repositories.book_repository import BookRepository

class BookService:
    @staticmethod
    def create_book(db: Session, book_data) -> Book:
        # 这里可以添加业务规则，如：检查同名图书是否存在
        return BookRepository.create(db, book_data)

    @staticmethod
    def list_books(db: Session) -> List[Book]:
        return BookRepository.get_all(db)
    
    @staticmethod
    def get_book(db: Session, book_id: int) -> Optional[Book]:
        return BookRepository.get(db, book_id)

6. 路由层（routers/book_router.py）

from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from services.book_service import BookService
from schemas.book_schema import BookCreate, BookResponse

# 创建路由实例，前缀为 /books
router = APIRouter(prefix="/books", tags=["Books"])

@router.post("/", response_model=BookResponse)
def create_book(book: BookCreate, db: Session = Depends(get_db)) -> BookResponse:
    """创建图书"""
    return BookService.create_book(db, book)

@router.get("/", response_model=List[BookResponse])
def list_books(db: Session = Depends(get_db)) -> List[BookResponse]:
    """获取图书列表"""
    return BookService.list_books(db)

@router.get("/{book_id}", response_model=BookResponse)
def get_book(book_id: int, db: Session = Depends(get_db)) -> BookResponse:
    """获取单本图书"""
    book = BookService.get_book(db, book_id)
    if not book:
        raise HTTPException(status_code=404, detail="Book not found")
    return book

7. 应用入口（main.py）

from fastapi import FastAPI
from database import engine
from models.book import Base
from routers.book_router import router as book_router

# 创建数据库表（实际项目建议使用 Alembic 做迁移）
Base.metadata.create_all(bind=engine)

# 创建 FastAPI 应用实例
app = FastAPI(title="图书管理系统")

# 注册路由
app.include_router(book_router)