一，sqlalchemy orm的核心概念

参考这里

二，创建项目目录和虚拟环境

# 1. 创建项目目录
mkdir test_project
cd test_project

# 2. 创建虚拟环境
python -m venv venv

# 3. 激活虚拟环境
# Windows:
venv\Scripts\activate
# Linux/Mac:
source venv/bin/activate

# 4. 创建基础目录结构
mkdir -p app/{api,core,crud,db,models,schemas}
mkdir alembic tests

三，创建 requirements.txt

# requirements.txt

alembic==1.17.2
annotated-doc==0.0.4
annotated-types==0.7.0
anyio==4.12.1
click==8.3.1
colorama==0.4.6
fastapi==0.128.0
greenlet==3.3.0
h11==0.16.0
httptools==0.7.1
idna==3.11
Mako==1.3.10
MarkupSafe==3.0.3
pydantic==2.12.5
pydantic-settings==2.12.0
pydantic_core==2.41.5
python-dotenv==1.2.1
PyYAML==6.0.3
SQLAlchemy==2.0.45
starlette==0.50.0
typing-inspection==0.4.2
typing_extensions==4.15.0
uvicorn==0.40.0
watchfiles==1.1.1
websockets==15.0.1

安装依赖：

pip install -r requirements.txt

四，创建环境配置文件

# .env

DATABASE_URL=sqlite:///./cookies.db
API_V1_STR=/api/v1
PROJECT_NAME=Test FastAPI Project

五，创建 .gitignore 文件

# .gitignore

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg

# Virtual Environment
venv/
env/
ENV/

# IDE
.vscode/
.idea/
*.swp
*.swo

# Database
*.db
*.sqlite3

# OS
.DS_Store
Thumbs.db

# Logs
*.log

# Alembic
alembic.ini

六，创建核心配置

# app/core/config.py
from typing import Optional
from pydantic_settings import BaseSettings
from pydantic import AnyHttpUrl, PostgresDsn, validator
from typing import List, Union


class Settings(BaseSettings):
    # 基础配置
    API_V1_STR: str = "/api/v1"
    PROJECT_NAME: str = "Test FastAPI Project"
    
    # 数据库配置
    DATABASE_URL: str = "sqlite:///./cookies.db"
    
    # 安全配置
    SECRET_KEY: str = "your-secret-key-change-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    
    # CORS 配置
    BACKEND_CORS_ORIGINS: List[AnyHttpUrl] = []
    
    @validator("BACKEND_CORS_ORIGINS", pre=True)
    def assemble_cors_origins(cls, v: Union[str, List[str]]) -> Union[List[str], str]:
        if isinstance(v, str) and not v.startswith("["):
            return [i.strip() for i in v.split(",")]
        elif isinstance(v, (list, str)):
            return v
        raise ValueError(v)
    
    # 模型配置
    class Config:
        case_sensitive = True
        env_file = ".env"


settings = Settings()

# app/core/__init__.py
# 空文件，用于标识包

七，创建数据库连接

# app/db/base.py
from sqlalchemy.ext.declarative import declarative_base

# 创建声明性基类
Base = declarative_base()

# app/db/session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from app.core.config import settings
from app.db.base import Base


# 创建数据库引擎
engine = create_engine(
    settings.DATABASE_URL,
    connect_args={"check_same_thread": False}  # SQLite 需要这个参数
)

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 依赖注入函数
def get_db():
    """
    获取数据库会话的依赖函数
    
    Yields:
        Session: SQLAlchemy 数据库会话
    
    使用示例：
        @app.get("/items")
        def read_items(db: Session = Depends(get_db)):
            return db.query(Item).all()
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# 辅助函数：创建所有表
def create_tables():
    """创建数据库中所有定义的表"""
    from app.models.cookie import Cookie  # 导入模型以确保表被注册
    Base.metadata.create_all(bind=engine)

# app/db/__init__.py
# 导出重要的数据库组件
from app.db.base import Base
from app.db.session import engine, SessionLocal, get_db, create_tables

__all__ = ["Base", "engine", "SessionLocal", "get_db", "create_tables"]

八，创建模型

# app/models/cookie.py
from datetime import datetime
from sqlalchemy import Column, Integer, Numeric, String, DateTime, Text
from sqlalchemy.sql import func

from app.db.base import Base


class Cookie(Base):
    """
    Cookie 数据模型
    
    对应数据库中的 cookies 表
    用于存储饼干产品信息
    """
    __tablename__ = "cookies"
    
    # 主键字段
    cookie_id = Column(
        Integer, 
        primary_key=True, 
        index=True,
        comment="饼干ID，主键"
    )
    
    # 基本字段
    cookie_name = Column(
        String(100), 
        nullable=False, 
        index=True,
        comment="饼干名称"
    )
    
    cookie_recipe_url = Column(
        String(500), 
        nullable=True,
        comment="饼干食谱URL"
    )
    
    cookie_sku = Column(
        String(100), 
        nullable=False, 
        unique=True,
        comment="饼干SKU（库存单位），唯一标识"
    )
    
    quantity = Column(
        Integer, 
        nullable=False, 
        default=0,
        comment="库存数量"
    )
    
    unit_cost = Column(
        Numeric(10, 2), 
        nullable=False,
        comment="单位成本（两位小数）"
    )
    
    description = Column(
        Text, 
        nullable=True,
        comment="饼干详细描述"
    )
    
    # 时间戳字段
    created_on = Column(
        DateTime(timezone=True), 
        server_default=func.now(),
        comment="创建时间"
    )
    
    updated_on = Column(
        DateTime(timezone=True), 
        server_default=func.now(), 
        onupdate=func.now(),
        comment="更新时间"
    )
    
    def __repr__(self):
        """对象的字符串表示"""
        return f"<Cookie(id={self.cookie_id}, name='{self.cookie_name}', sku='{self.cookie_sku}')>"
    
    def to_dict(self):
        """将模型转换为字典（便于JSON序列化）"""
        return {
            "cookie_id": self.cookie_id,
            "cookie_name": self.cookie_name,
            "cookie_recipe_url": self.cookie_recipe_url,
            "cookie_sku": self.cookie_sku,
            "quantity": self.quantity,
            "unit_cost": float(self.unit_cost) if self.unit_cost else None,
            "description": self.description,
            "created_on": self.created_on.isoformat() if self.created_on else None,
            "updated_on": self.updated_on.isoformat() if self.updated_on else None,
        }

# app/models/__init__.py
# 导出所有模型
from app.models.cookie import Cookie

__all__ = ["Cookie"]

九，创建 Pydantic 模式

# app/schemas/cookie.py
from typing import Optional
from datetime import datetime
from pydantic import BaseModel, Field, HttpUrl


class CookieBase(BaseModel):
    """
    Cookie 基础模式
    
    包含创建和更新 Cookie 时的共享字段
    """
    cookie_name: str = Field(
        ...,
        min_length=1,
        max_length=100,
        description="饼干名称，必填，1-100字符",
        example="巧克力曲奇"
    )
    
    cookie_recipe_url: Optional[HttpUrl] = Field(
        None,
        description="饼干食谱URL，可选",
        example="https://example.com/recipe/chocolate-cookie"
    )
    
    cookie_sku: str = Field(
        ...,
        min_length=1,
        max_length=100,
        description="饼干SKU，必填，唯一标识，1-100字符",
        example="COOKIE-CHOCO-001"
    )
    
    quantity: int = Field(
        0,
        ge=0,
        description="库存数量，默认为0，必须大于等于0",
        example=100
    )
    
    unit_cost: float = Field(
        ...,
        ge=0,
        description="单位成本，必填，必须大于等于0，两位小数",
        example=2.50
    )
    
    description: Optional[str] = Field(
        None,
        max_length=1000,
        description="饼干描述，可选，最多1000字符",
        example="美味的巧克力曲奇，外酥内软"
    )


class CookieCreate(CookieBase):
    """
    创建 Cookie 请求模式
    
    继承自 CookieBase，用于创建新 Cookie
    """
    pass


class CookieUpdate(BaseModel):
    """
    更新 Cookie 请求模式
    
    用于更新现有 Cookie，所有字段都是可选的
    """
    cookie_name: Optional[str] = Field(
        None,
        min_length=1,
        max_length=100,
        description="饼干名称，可选，1-100字符"
    )
    
    cookie_recipe_url: Optional[HttpUrl] = Field(
        None,
        description="饼干食谱URL，可选"
    )
    
    cookie_sku: Optional[str] = Field(
        None,
        min_length=1,
        max_length=100,
        description="饼干SKU，可选，唯一标识，1-100字符"
    )
    
    quantity: Optional[int] = Field(
        None,
        ge=0,
        description="库存数量，可选，必须大于等于0"
    )
    
    unit_cost: Optional[float] = Field(
        None,
        ge=0,
        description="单位成本，可选，必须大于等于0，两位小数"
    )
    
    description: Optional[str] = Field(
        None,
        max_length=1000,
        description="饼干描述，可选，最多1000字符"
    )


class CookieInDB(CookieBase):
    """
    数据库中的 Cookie 模式
    
    包含从数据库读取的所有字段
    """
    cookie_id: int = Field(..., description="饼干ID")
    created_on: datetime = Field(..., description="创建时间")
    updated_on: datetime = Field(..., description="更新时间")
    
    class Config:
        from_attributes = True  # 支持从ORM对象转换


class CookieResponse(BaseModel):
    """
    Cookie 响应模式
    
    标准的API响应格式
    """
    code: int = Field(default=200, description="状态码")
    msg: str = Field(default="success", description="消息")
    data: CookieInDB = Field(..., description="饼干数据")


# 列表响应相关模式
class CookieListData(BaseModel):
    """Cookie 列表数据模式"""
    items: list[CookieInDB] = Field(..., description="饼干列表")
    total: int = Field(..., description="总记录数")
    page: int = Field(..., description="当前页码")
    page_size: int = Field(..., description="每页数量")
    total_pages: Optional[int] = Field(None, description="总页数")


class CookieListResponse(BaseModel):
    """Cookie 列表响应模式"""
    code: int = Field(default=200, description="状态码")
    msg: str = Field(default="success", description="消息")
    data: CookieListData = Field(..., description="分页数据")

# app/schemas/__init__.py
# 导出所有模式
from app.schemas.cookie import (
    CookieBase, CookieCreate, CookieUpdate, 
    CookieInDB, CookieResponse, CookieListResponse, CookieListData
)

__all__ = [
    "CookieBase", "CookieCreate", "CookieUpdate",
    "CookieInDB", "CookieResponse", "CookieListResponse", "CookieListData"
]

十，创建 CRUD 操作

# app/crud/cookie.py
from typing import Optional, List, Dict, Any
from sqlalchemy.orm import Session
from sqlalchemy import desc, asc

from app.models.cookie import Cookie
from app.schemas.cookie import CookieCreate, CookieUpdate


class CRUDCookie:
    """
    Cookie 的 CRUD 操作类
    
    提供对 cookies 表的增删改查操作
    """
    
    def get(self, db: Session, cookie_id: int) -> Optional[Cookie]:
        """
        根据ID获取单个 Cookie
        
        Args:
            db: 数据库会话
            cookie_id: 饼干ID
        
        Returns:
            Cookie 对象或 None
        """
        return db.query(Cookie).filter(Cookie.cookie_id == cookie_id).first()
    
    def get_by_sku(self, db: Session, sku: str) -> Optional[Cookie]:
        """
        根据SKU获取 Cookie
        
        Args:
            db: 数据库会话
            sku: 饼干SKU
        
        Returns:
            Cookie 对象或 None
        """
        return db.query(Cookie).filter(Cookie.cookie_sku == sku).first()
    
    def get_multi(
        self, 
        db: Session, 
        skip: int = 0, 
        limit: int = 100,
        sort_by: str = "cookie_id",
        sort_order: str = "asc"
    ) -> List[Cookie]:
        """
        获取多个 Cookie（支持分页和排序）
        
        Args:
            db: 数据库会话
            skip: 跳过的记录数
            limit: 返回的最大记录数
            sort_by: 排序字段
            sort_order: 排序顺序（asc 或 desc）
        
        Returns:
            Cookie 对象列表
        """
        # 构建查询
        query = db.query(Cookie)
        
        # 应用排序
        if hasattr(Cookie, sort_by):
            column = getattr(Cookie, sort_by)
            if sort_order.lower() == "desc":
                query = query.order_by(desc(column))
            else:
                query = query.order_by(asc(column))
        else:
            # 默认按ID升序排序
            query = query.order_by(Cookie.cookie_id)
        
        # 应用分页
        return query.offset(skip).limit(limit).all()
    
    def get_multi_by_filter(
        self,
        db: Session,
        skip: int = 0,
        limit: int = 100,
        **filters: Dict[str, Any]
    ) -> List[Cookie]:
        """
        根据过滤条件获取多个 Cookie
        
        Args:
            db: 数据库会话
            skip: 跳过的记录数
            limit: 返回的最大记录数
            **filters: 过滤条件，如 cookie_name="xxx"
        
        Returns:
            过滤后的 Cookie 对象列表
        """
        query = db.query(Cookie)
        
        # 应用过滤条件
        for field, value in filters.items():
            if hasattr(Cookie, field) and value is not None:
                column = getattr(Cookie, field)
                if isinstance(value, str):
                    # 字符串字段使用模糊搜索
                    query = query.filter(column.ilike(f"%{value}%"))
                else:
                    query = query.filter(column == value)
        
        return query.offset(skip).limit(limit).all()
    
    def create(self, db: Session, obj_in: CookieCreate) -> Cookie:
        """
        创建新的 Cookie
        
        Args:
            db: 数据库会话
            obj_in: Cookie 创建数据
        
        Returns:
            创建的 Cookie 对象
        
        Raises:
            ValueError: 如果SKU已存在
        """
        # 检查SKU是否已存在
        existing = self.get_by_sku(db, obj_in.cookie_sku)
        if existing:
            raise ValueError(f"SKU '{obj_in.cookie_sku}' 已存在")
        
        # 创建模型实例
        db_obj = Cookie(
            cookie_name=obj_in.cookie_name,
            cookie_recipe_url=str(obj_in.cookie_recipe_url) if obj_in.cookie_recipe_url else None,
            cookie_sku=obj_in.cookie_sku,
            quantity=obj_in.quantity,
            unit_cost=obj_in.unit_cost,
            description=obj_in.description
        )
        
        # 保存到数据库
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj
    
    def update(
        self, 
        db: Session, 
        db_obj: Cookie, 
        obj_in: CookieUpdate
    ) -> Cookie:
        """
        更新现有的 Cookie
        
        Args:
            db: 数据库会话
            db_obj: 要更新的 Cookie 对象
            obj_in: 更新数据
        
        Returns:
            更新后的 Cookie 对象
        """
        update_data = obj_in.model_dump(exclude_unset=True)
        
        # 如果要更新SKU，检查是否重复
        if "cookie_sku" in update_data and update_data["cookie_sku"] != db_obj.cookie_sku:
            existing = self.get_by_sku(db, update_data["cookie_sku"])
            if existing and existing.cookie_id != db_obj.cookie_id:
                raise ValueError(f"SKU '{update_data['cookie_sku']}' 已存在")
        
        # 更新字段
        for field, value in update_data.items():
            if hasattr(db_obj, field):
                if field == "cookie_recipe_url" and value:
                    setattr(db_obj, field, str(value))
                else:
                    setattr(db_obj, field, value)
        
        # 保存更改
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj
    
    def remove(self, db: Session, cookie_id: int) -> Cookie:
        """
        删除 Cookie
        
        Args:
            db: 数据库会话
            cookie_id: 要删除的饼干ID
        
        Returns:
            被删除的 Cookie 对象
        
        Raises:
            ValueError: 如果找不到指定的 Cookie
        """
        obj = self.get(db, cookie_id)
        if not obj:
            raise ValueError(f"Cookie ID {cookie_id} 不存在")
        
        db.delete(obj)
        db.commit()
        return obj
    
    def get_total_count(self, db: Session) -> int:
        """
        获取 Cookie 总数
        
        Args:
            db: 数据库会话
        
        Returns:
            总记录数
        """
        return db.query(Cookie).count()


# 创建 CRUDCookie 实例
cookie = CRUDCookie()

# app/crud/__init__.py
# 导出所有 CRUD 实例
from app.crud.cookie import cookie

__all__ = ["cookie"]

十一，创建 API 路由

# app/api/api_v1/api.py
from fastapi import APIRouter

# 导入各个端点的路由
from app.api.api_v1.endpoints import cookies

# 创建 API v1 路由器
api_router = APIRouter()

# 注册各个端点的路由
api_router.include_router(cookies.router, prefix="/cookies", tags=["cookies"])

# app/api/api_v1/endpoints/cookies.py
from typing import Optional, List
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session

from app.db.session import get_db
from app.crud import cookie as crud_cookie
from app.schemas.cookie import (
    CookieCreate, CookieUpdate, CookieResponse, 
    CookieListResponse, CookieInDB
)

# 创建路由器
router = APIRouter()


@router.get("/", response_model=CookieListResponse)
def read_cookies(
    db: Session = Depends(get_db),
    page: int = Query(1, ge=1, description="页码，从1开始"),
    page_size: int = Query(10, ge=1, le=100, description="每页数量，最大100"),
    sort_by: str = Query("cookie_id", description="排序字段"),
    sort_order: str = Query("asc", description="排序顺序，asc或desc"),
    name: Optional[str] = Query(None, description="按名称搜索"),
    sku: Optional[str] = Query(None, description="按SKU搜索"),
):
    """
    获取饼干列表（支持分页、排序和搜索）
    
    - **page**: 页码，默认为1
    - **page_size**: 每页数量，默认为10，最大100
    - **sort_by**: 排序字段，默认为cookie_id
    - **sort_order**: 排序顺序，默认为asc（升序）
    - **name**: 按饼干名称搜索（模糊匹配）
    - **sku**: 按饼干SKU搜索（精确匹配）
    """
    try:
        # 计算跳过的记录数
        skip = (page - 1) * page_size
        
        # 获取数据
        if name or sku:
            # 使用过滤查询
            filters = {}
            if name:
                filters["cookie_name"] = name
            if sku:
                filters["cookie_sku"] = sku
            
            cookies = crud_cookie.get_multi_by_filter(
                db, skip=skip, limit=page_size, **filters
            )
            total = len(cookies)  # 简化处理，实际应查询总数
        else:
            # 使用普通分页查询
            cookies = crud_cookie.get_multi(
                db, skip=skip, limit=page_size, 
                sort_by=sort_by, sort_order=sort_order
            )
            total = crud_cookie.get_total_count(db)
        
        # 计算总页数
        total_pages = (total + page_size - 1) // page_size if page_size > 0 else 0
        
        return CookieListResponse(
            code=200,
            msg="success",
            data={
                "items": cookies,
                "total": total,
                "page": page,
                "page_size": page_size,
                "total_pages": total_pages
            }
        )
        
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取数据失败: {str(e)}"
        )


@router.get("/{cookie_id}", response_model=CookieResponse)
def read_cookie(
    cookie_id: int, 
    db: Session = Depends(get_db)
):
    """
    根据ID获取单个饼干信息
    
    - **cookie_id**: 饼干ID（路径参数）
    """
    db_cookie = crud_cookie.get(db, cookie_id=cookie_id)
    if db_cookie is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"饼干ID {cookie_id} 不存在"
        )
    
    return CookieResponse(
        code=200,
        msg="success",
        data=db_cookie
    )


@router.post("/", response_model=CookieResponse, status_code=status.HTTP_201_CREATED)
def create_cookie(
    cookie_in: CookieCreate, 
    db: Session = Depends(get_db)
):
    """
    创建新的饼干
    
    - **cookie_name**: 饼干名称（必填）
    - **cookie_recipe_url**: 食谱URL（可选）
    - **cookie_sku**: 饼干SKU（必填，唯一）
    - **quantity**: 库存数量（可选，默认0）
    - **unit_cost**: 单位成本（必填）
    - **description**: 描述（可选）
    """
    try:
        db_cookie = crud_cookie.create(db, obj_in=cookie_in)
        
        return CookieResponse(
            code=201,
            msg="创建成功",
            data=db_cookie
        )
        
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e)
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"创建失败: {str(e)}"
        )


@router.put("/{cookie_id}", response_model=CookieResponse)
def update_cookie(
    cookie_id: int, 
    cookie_in: CookieUpdate, 
    db: Session = Depends(get_db)
):
    """
    更新饼干信息
    
    - **cookie_id**: 要更新的饼干ID（路径参数）
    - 其他字段：要更新的数据（所有字段都是可选的）
    """
    db_cookie = crud_cookie.get(db, cookie_id=cookie_id)
    if db_cookie is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"饼干ID {cookie_id} 不存在"
        )
    
    try:
        db_cookie = crud_cookie.update(db, db_obj=db_cookie, obj_in=cookie_in)
        
        return CookieResponse(
            code=200,
            msg="更新成功",
            data=db_cookie
        )
        
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e)
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"更新失败: {str(e)}"
        )


@router.delete("/{cookie_id}", response_model=dict)
def delete_cookie(
    cookie_id: int, 
    db: Session = Depends(get_db)
):
    """
    删除饼干
    
    - **cookie_id**: 要删除的饼干ID（路径参数）
    """
    db_cookie = crud_cookie.get(db, cookie_id=cookie_id)
    if db_cookie is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"饼干ID {cookie_id} 不存在"
        )
    
    try:
        crud_cookie.remove(db, cookie_id=cookie_id)
        
        return {
            "code": 200,
            "msg": "删除成功",
            "data": None
        }
        
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"删除失败: {str(e)}"
        )


@router.get("/sku/{sku}", response_model=CookieResponse)
def read_cookie_by_sku(
    sku: str, 
    db: Session = Depends(get_db)
):
    """
    根据SKU获取饼干信息
    
    - **sku**: 饼干SKU（路径参数）
    """
    db_cookie = crud_cookie.get_by_sku(db, sku=sku)
    if db_cookie is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"饼干SKU '{sku}' 不存在"
        )
    
    return CookieResponse(
        code=200,
        msg="success",
        data=db_cookie
    )

# app/api/__init__.py
# 空文件，用于标识包

# app/api/api_v1/__init__.py
# 空文件，用于标识包

# app/api/api_v1/endpoints/__init__.py
# 导出所有端点路由
from app.api.api_v1.endpoints.cookies import router as cookies_router

__all__ = ["cookies_router"]

十二，创建 FastAPI 应用主文件

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager

from app.core.config import settings
from app.api.api_v1.api import api_router
from app.db.session import create_tables


@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    应用生命周期管理
    
    - 启动时：创建数据库表
    - 关闭时：清理资源
    """
    # 启动时执行
    print("正在启动应用...")
    create_tables()
    print("数据库表已创建/检查完成")
    
    yield  # 应用运行期间
    
    # 关闭时执行
    print("正在关闭应用...")


# 创建 FastAPI 应用实例
app = FastAPI(
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.API_V1_STR}/openapi.json",
    version="1.0.0",
    description="""
    ## Cookie 管理 API
    
    一个用于管理饼干产品的 RESTful API，提供完整的 CRUD 操作。
    
    ### 功能特性
    
    - 饼干的增删改查（CRUD）
    - 分页和排序支持
    - 搜索和过滤功能
    - 数据验证和错误处理
    - OpenAPI 文档自动生成
    
    ### 使用说明
    
    1. 访问 `/docs` 查看交互式 API 文档
    2. 使用 `/api/v1/cookies/` 端点管理饼干数据
    3. 所有请求和响应都使用 JSON 格式
    """,
    contact={
        "name": "开发者",
        "email": "developer@example.com",
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT",
    },
    lifespan=lifespan,
)

# 设置 CORS
if settings.BACKEND_CORS_ORIGINS:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

# 注册 API 路由
app.include_router(api_router, prefix=settings.API_V1_STR)


# 根路由
@app.get("/", tags=["根"])
async def root():
    """
    根端点
    
    返回应用信息和可用端点
    """
    return {
        "app": settings.PROJECT_NAME,
        "version": "1.0.0",
        "docs": "/docs",
        "redoc": "/redoc",
        "api_v1": settings.API_V1_STR,
        "endpoints": {
            "cookies": f"{settings.API_V1_STR}/cookies",
        }
    }


# 健康检查端点
@app.get("/health", tags=["健康检查"])
async def health_check():
    """
    健康检查端点
    
    用于监控应用运行状态
    """
    return {
        "status": "healthy",
        "timestamp": "2024-01-01T00:00:00Z"  # 实际应该使用 datetime.now()
    }


if __name__ == "__main__":
    import uvicorn
    
    # 开发服务器配置
    uvicorn.run(
        "app.main:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="info"
    )

十三，创建 Alembic 迁移配置

# 初始化 Alembic
alembic init alembic

修改 alembic.ini：

sqlalchemy.url = sqlite:///./cookies.db

修改 alembic/env.py：

import os
import sys
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context

# 添加项目根目录到 Python 路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# 导入模型
from app.db.base import Base
target_metadata = Base.metadata

# 其余配置保持不变，但在 configure 中添加 SQLite 支持
def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            render_as_batch=True,  # SQLite 需要这个
            compare_type=True,
        )

        with context.begin_transaction():
            context.run_migrations()

创建初始迁移：

alembic revision --autogenerate -m "initial migration"
alembic upgrade head

十四，创建测试数据脚本

# scripts/create_test_data.py
#!/usr/bin/env python3
"""
测试数据生成脚本

用于向数据库中添加测试用的饼干数据
"""
import sys
import os

# 添加项目根目录到 Python 路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy.orm import Session
from app.db.session import SessionLocal
from app.crud import cookie as crud_cookie
from app.schemas.cookie import CookieCreate


def create_test_cookies(db: Session):
    """创建测试饼干数据"""
    test_cookies = [
        CookieCreate(
            cookie_name="巧克力曲奇",
            cookie_recipe_url="https://example.com/recipes/chocolate-chip",
            cookie_sku="COOKIE-CHOCO-001",
            quantity=100,
            unit_cost=2.50,
            description="香浓巧克力曲奇，外酥内软"
        ),
        CookieCreate(
            cookie_name="燕麦葡萄干曲奇",
            cookie_recipe_url="https://example.com/recipes/oatmeal-raisin",
            cookie_sku="COOKIE-OAT-002",
            quantity=75,
            unit_cost=2.00,
            description="健康燕麦与甜葡萄干的完美结合"
        ),
        CookieCreate(
            cookie_name="花生酱曲奇",
            cookie_recipe_url="https://example.com/recipes/peanut-butter",
            cookie_sku="COOKIE-PB-003",
            quantity=120,
            unit_cost=2.25,
            description="浓郁花生酱风味，口感丰富"
        ),
        CookieCreate(
            cookie_name="奶油曲奇",
            cookie_sku="COOKIE-BUTTER-004",
            quantity=90,
            unit_cost=1.80,
            description="经典奶油曲奇，入口即化"
        ),
        CookieCreate(
            cookie_name="抹茶曲奇",
            cookie_recipe_url="https://example.com/recipes/matcha",
            cookie_sku="COOKIE-MATCHA-005",
            quantity=60,
            unit_cost=3.00,
            description="日式抹茶风味，清新不腻"
        ),
    ]
    
    created_count = 0
    for cookie_data in test_cookies:
        try:
            crud_cookie.create(db, obj_in=cookie_data)
            created_count += 1
            print(f"创建饼干: {cookie_data.cookie_name}")
        except ValueError as e:
            print(f"跳过已存在的饼干: {cookie_data.cookie_name} ({str(e)})")
    
    return created_count


def main():
    """主函数"""
    print("开始创建测试数据...")
    
    db = SessionLocal()
    try:
        created = create_test_cookies(db)
        print(f"\n成功创建 {created} 个测试饼干")
        
        # 显示总数
        total = crud_cookie.get_total_count(db)
        print(f"数据库中共有 {total} 个饼干")
        
    except Exception as e:
        print(f"创建测试数据时出错: {str(e)}")
        db.rollback()
    finally:
        db.close()
    
    print("\n测试数据创建完成！")


if __name__ == "__main__":
    main()

十五，项目结构

test_project/
├── alembic/
│   ├── versions/
│   ├── env.py
│   ├── README
│   └── script.py.mako
├── app/
│   ├── api/
│   │   └── api_v1/
│   │       ├── endpoints/
│   │       │   ├── __init__.py
│   │       │   └── cookies.py
│   │       ├── __init__.py
│   │       └── api.py
│   ├── core/
│   │   ├── __init__.py
│   │   └── config.py
│   ├── crud/
│   │   ├── __init__.py
│   │   └── cookie.py
│   ├── db/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   └── session.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── cookie.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── cookie.py
│   └── main.py
├── scripts/
│   └── create_test_data.py
├── .env
├── .env.example
├── .gitignore
├── alembic.ini
├── cookies.db
├── requirements.txt
├── README.md
├── run.sh
└── run.bat

十五，总结与问题

（一）app/api/api_v1/endpoints/cookies.py的区别app/crud/cookie.py

✅ 应该放在 endpoints/cookies.py：

• HTTP 相关的一切：路由定义、请求/响应处理
• 输入验证：路径参数、查询参数、请求体
• 错误处理：HTTP 异常、状态码
• 响应构建：格式化响应数据
• 认证授权：权限检查
• 调用 CRUD 层：不直接操作数据库

✅ 应该放在 crud/cookie.py：

• 数据库操作：所有 SQL 查询
• ORM 操作：SQLAlchemy 相关代码
• 事务管理：commit、rollback
• 查询构建：复杂的查询逻辑
• 业务逻辑：数据层面的验证和逻辑
• 返回 ORM 对象：不处理 HTTP 响应

这种分层架构的好处：

• 关注点分离：API 层只关心 HTTP，CRUD 层只关心数据库
• 可测试性：可以单独测试 API 逻辑和数据库逻辑
• 可维护性：代码职责清晰，易于维护和扩展
• 可重用性：CRUD 层可以被不同的 API 端点复用