初次代码

.gitignore

@@ -0,0 +1,14 @@
+# 忽略虚拟环境（依赖包很大，别人自己本地安装即可）
+venv/
+.venv/
+env/
+
+# 忽略 PyCharm 自身的配置文件
+.idea/
+
+# 忽略 Python 自动生成的运行缓存
+__pycache__/
+*.pyc
+
+# 忽略本地环境变量文件（防止泄露云数据库密码！）
+.env

api1/users.py

@@ -0,0 +1,98 @@
+# api/users.py
+# 本文件定义用户相关的所有 API 路由（Controller 层）
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+from typing import List
+
+from database import get_db
+from dao.users_dao import UserDAO
+from scheme.users import UserCreate, UserUpdate, UserResponse
+
+# 创建路由器，前缀将在 main0814.py 中统一添加
+router = APIRouter()
+
+# ---------- 查询所有用户（分页） ----------
+@router.get("/", response_model=List[UserResponse])
+async def get_users(
+    skip: int = Query(0, ge=0, description="跳过的记录数"),
+    limit: int = Query(100, ge=1, le=200, description="返回的最大记录数"),
+    db: Session = Depends(get_db)   # 依赖注入获得数据库会话
+):
+    """
+    获取用户列表，支持分页
+    """
+    users = UserDAO.get_all(db, skip=skip, limit=limit)
+    return users   # FastAPI 自动根据 response_model 转换为 JSON
+
+# ---------- 根据 ID 查询单个用户 ----------
+@router.get("/{user_id}", response_model=UserResponse)
+async def get_user(
+    user_id: int,
+    db: Session = Depends(get_db)
+):
+    """
+    根据用户 ID 获取详细信息
+    """
+    user = UserDAO.get_by_id(db, user_id)
+    if not user:
+        raise HTTPException(status_code=404, detail="用户不存在")
+    return user
+
+# ---------- 创建新用户 ----------
+@router.post("/", response_model=UserResponse, status_code=201)
+async def create_user(
+    user_data: UserCreate,
+    db: Session = Depends(get_db)
+):
+    """
+    创建新用户，需要提供用户名、邮箱和全名（全名可选）
+    注意：用户名必须唯一
+    """
+    # 检查用户名是否已被占用
+    existing = UserDAO.get_by_username(db, user_data.username)
+    if existing:
+        raise HTTPException(status_code=400, detail="用户名已被占用")
+    # 调用 DAO 创建用户
+    new_user = UserDAO.create(db, user_data)
+    return new_user
+
+# ---------- 更新用户信息 ----------
+@router.put("/{user_id}", response_model=UserResponse)
+async def update_user(
+    user_id: int,
+    user_data: UserUpdate,
+    db: Session = Depends(get_db)
+):
+    """
+    更新用户信息（只更新传入的字段）
+    """
+    # 检查用户是否存在
+    existing = UserDAO.get_by_id(db, user_id)
+    if not existing:
+        raise HTTPException(status_code=404, detail="用户不存在")
+
+    # 如果更新用户名，需要检查新用户名是否与其他用户冲突（排除自身）
+    if user_data.username is not None:
+        conflict = UserDAO.get_by_username(db, user_data.username)
+        if conflict and conflict.id != user_id:
+            raise HTTPException(status_code=400, detail="用户名已被其他用户占用")
+
+    # 执行更新
+    updated = UserDAO.update(db, user_id, user_data)
+    return updated
+
+# ---------- 删除用户 ----------
+@router.delete("/{user_id}", status_code=204)
+async def delete_user(
+    user_id: int,
+    db: Session = Depends(get_db)
+):
+    """
+    删除用户，成功返回 204 No Content
+    """
+    success = UserDAO.delete(db, user_id)
+    if not success:
+        raise HTTPException(status_code=404, detail="用户不存在")
+    # 返回 None 表示 204 状态码（无内容）
+    return None

dao/users_dao.py

@@ -0,0 +1,86 @@
+# dao/users_dao.py
+# 本文件封装对 User 表的所有数据库操作（增、删、改、查）
+
+from sqlalchemy.orm import Session
+from model.users import User
+from scheme.users import UserCreate, UserUpdate
+from typing import Optional, List
+
+class UserDAO:
+    """用户数据访问对象，所有方法均为静态方法，方便调用"""
+
+    @staticmethod
+    def get_all(db: Session, skip: int = 0, limit: int = 100) -> List[User]:
+        """
+        获取所有用户（支持分页）
+        :param db: 数据库会话
+        :param skip: 偏移量（跳过前 skip 条）
+        :param limit: 最大返回条数
+        :return: 用户对象列表
+        """
+        return db.query(User).offset(skip).limit(limit).all()
+
+    @staticmethod
+    def get_by_id(db: Session, user_id: int) -> Optional[User]:
+        """
+        根据主键 ID 获取单个用户
+        :return: 用户对象或 None
+        """
+        return db.query(User).filter(User.id == user_id).first()
+
+    @staticmethod
+    def get_by_username(db: Session, username: str) -> Optional[User]:
+        """
+        根据用户名获取用户（用于唯一性检查）
+        """
+        return db.query(User).filter(User.username == username).first()
+
+    @staticmethod
+    def create(db: Session, user_data: UserCreate) -> User:
+        """
+        创建新用户
+        :param db: 数据库会话
+        :param user_data: 符合 UserCreate 模型的数据
+        :return: 创建后的 User 对象（含自增 id 和默认时间）
+        """
+        # 将 Pydantic 模型转为字典，并解包构建 SQLAlchemy 模型实例
+        db_user = User(**user_data.model_dump())
+        db.add(db_user)        # 添加到会话
+        db.commit()            # 提交事务，此时会执行 INSERT，并自动填充自增字段
+        db.refresh(db_user)    # 刷新对象，获取数据库生成的默认值（如 created_at）
+        return db_user
+
+    @staticmethod
+    def update(db: Session, user_id: int, user_data: UserUpdate) -> Optional[User]:
+        """
+        更新用户信息（只更新传入的非空字段）
+        :param db: 数据库会话
+        :param user_id: 要更新的用户 ID
+        :param user_data: 包含要更新字段的 Pydantic 模型
+        :return: 更新后的 User 对象，如果用户不存在则返回 None
+        """
+        db_user = UserDAO.get_by_id(db, user_id)
+        if not db_user:
+            return None
+
+        # 只更新客户端显式传入的字段（exclude_unset=True 排除未设置的字段）
+        update_data = user_data.model_dump(exclude_unset=True)
+        for key, value in update_data.items():
+            setattr(db_user, key, value)   # 动态设置属性
+
+        db.commit()        # 提交事务
+        db.refresh(db_user)  # 刷新对象，获取 onupdate 时间等
+        return db_user
+
+    @staticmethod
+    def delete(db: Session, user_id: int) -> bool:
+        """
+        删除用户
+        :return: True 表示删除成功，False 表示用户不存在
+        """
+        db_user = UserDAO.get_by_id(db, user_id)
+        if not db_user:
+            return False
+        db.delete(db_user)   # 标记删除
+        db.commit()          # 提交事务
+        return True

database.py

@@ -0,0 +1,39 @@
+# database.py
+# 本文件负责配置数据库连接、创建引擎、会话工厂，并提供依赖注入函数
+
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+# 1. 配置 MySQL 数据库连接 URL
+# 格式：mysql+pymysql://用户名:密码@主机:端口/数据库名?编码
+# 请将下面的 'root', '123456', 'localhost', '3306', 'test_db' 替换为你自己的实际信息
+SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/test_db"
+
+# 2. 创建数据库引擎
+#    - pool_pre_ping=True 表示每次从连接池取出连接前先 ping 一下，防止使用已断开的连接
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL,
+    pool_pre_ping=True,
+    echo=True  # 设置为 True 会在控制台打印所有 SQL 语句，便于调试，生产环境可关闭
+)
+
+# 3. 创建会话工厂
+#    autocommit=False：不自动提交，需要手动 commit
+#    autoflush=False：不自动 flush，在查询前会自动 flush，一般保持默认
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+# 4. 创建声明式基类，所有模型类都继承自它
+Base = declarative_base()
+
+# 5. 依赖注入函数：用于 FastAPI 路由中获取数据库会话
+def get_db():
+    """
+    每次请求创建一个数据库会话，请求结束后关闭。
+    这个函数会作为 Depends 的参数注入到路由中。
+    """
+    db = SessionLocal()          # 创建一个会话实例
+    try:
+        yield db                 # 将会话交给路由函数使用
+    finally:
+        db.close()               # 无论是否发生异常，最后都会关闭会话，释放连接

main6.py

@@ -0,0 +1,48 @@
+# main0814.py
+# 项目入口文件：创建 FastAPI 应用、注册路由、创建数据库表
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from database import engine, Base
+from api1 import users   # 导入 users 子路由
+
+# 1. 创建数据库表（如果表不存在）
+#    Base.metadata.create_all 会扫描所有继承 Base 的模型，生成对应的 CREATE TABLE 语句
+#    对于 MySQL，它会根据模型定义自动生成建表语句（包括索引、约束等）
+# Base.metadata.create_all(bind=engine)
+
+# 2. 创建 FastAPI 实例
+app = FastAPI(
+    title="FastAPI + SQLAlchemy 分层架构（MySQL）",
+    description="用户管理示例，演示分层架构和 MySQL 集成",
+    version="1.0.0"
+)
+
+# 3. 添加跨域中间件（允许前端跨域请求）
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],           # 允许所有来源，生产环境应指定具体域名
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 4. 注册子路由
+#    prefix 为路由前缀，所有用户接口都以 /api/users 开头
+app.include_router(users.router, prefix="/api/users", tags=["用户管理"])
+
+# 5. 根路径
+@app.get("/")
+async def root():
+    return {"message": "欢迎访问 FastAPI + SQLAlchemy 分层示例！请访问 /docs 查看 API 文档。"}
+
+# 6. 如果直接运行此文件，启动 uvicorn 服务器
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(
+        "main6:app",      # 指定应用位置（模块名:应用变量名）
+        host="localhost",  # 监听所有网络接口
+        port=8001,       # 端口
+        reload=True      # 开发模式，代码变动自动重启
+    )

model/users.py

@@ -0,0 +1,26 @@
+# model/users.py
+# 本文件定义 User 表的结构，映射到 MySQL 数据库
+
+from sqlalchemy import Column, Integer, String, DateTime
+from sqlalchemy.sql import func
+from database import Base
+
+class User(Base):
+    """
+    用户表模型
+    对应 MySQL 中的 users 表
+    """
+    __tablename__ = "users"  # 表名
+
+    # 字段定义
+    id = Column(Integer, primary_key=True, index=True)          # 主键，自增，索引
+    username = Column(String(50), unique=True, index=True, nullable=False)  # 用户名，唯一，非空
+    email = Column(String(100), unique=True, index=True, nullable=False)    # 邮箱，唯一，非空
+    full_name = Column(String(100), nullable=True)              # 全名，可为空
+
+    # created_at：创建时间，自动设置为当前时间（服务器时间）
+    # server_default=func.now() 表示由数据库生成默认值
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+
+    # updated_at：更新时间，当记录更新时自动设置为当前时间（由 SQLAlchemy 的 onupdate 触发）
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

scheme/users.py

@@ -0,0 +1,27 @@
+# scheme/users.py
+from pydantic import BaseModel, Field, EmailStr
+from datetime import datetime
+from typing import Optional
+
+# ---------- 请求模型 ----------
+class UserCreate(BaseModel):
+    username: str = Field(..., min_length=3, max_length=50)
+    email: EmailStr
+    full_name: Optional[str] = Field(None, max_length=100)
+
+class UserUpdate(BaseModel):
+    username: Optional[str] = Field(None, min_length=3, max_length=50)
+    email: Optional[EmailStr] = None
+    full_name: Optional[str] = Field(None, max_length=100)
+
+# ---------- 响应模型 ----------
+class UserResponse(BaseModel):
+    id: int
+    username: str
+    email: str
+    full_name: Optional[str]
+    created_at: datetime
+    updated_at: Optional[datetime]
+
+    class Config:
+        from_attributes = True  # 支持 ORM 对象转换