From cea442451db171d9cf3108d2aac84f5f7c6b4e07 Mon Sep 17 00:00:00 2001 From: jianqi Date: Fri, 11 Sep 2026 23:40:12 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=9D=E6=AC=A1=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 14 +++++++ api1/__init__.py | 0 api1/users.py | 98 ++++++++++++++++++++++++++++++++++++++++++++++ dao/__init__.py | 0 dao/users_dao.py | 86 ++++++++++++++++++++++++++++++++++++++++ database.py | 39 ++++++++++++++++++ main6.py | 48 +++++++++++++++++++++++ model/__init__.py | 0 model/users.py | 26 ++++++++++++ scheme/__init__.py | 0 scheme/users.py | 27 +++++++++++++ 11 files changed, 338 insertions(+) create mode 100644 .gitignore create mode 100644 api1/__init__.py create mode 100644 api1/users.py create mode 100644 dao/__init__.py create mode 100644 dao/users_dao.py create mode 100644 database.py create mode 100644 main6.py create mode 100644 model/__init__.py create mode 100644 model/users.py create mode 100644 scheme/__init__.py create mode 100644 scheme/users.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8f41b71 --- /dev/null +++ b/.gitignore @@ -0,0 +1,14 @@ +# 忽略虚拟环境（依赖包很大，别人自己本地安装即可） +venv/ +.venv/ +env/ + +# 忽略 PyCharm 自身的配置文件 +.idea/ + +# 忽略 Python 自动生成的运行缓存 +__pycache__/ +*.pyc + +# 忽略本地环境变量文件（防止泄露云数据库密码！） +.env \ No newline at end of file diff --git a/api1/__init__.py b/api1/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api1/users.py b/api1/users.py new file mode 100644 index 0000000..afdcf75 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/dao/__init__.py b/dao/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/dao/users_dao.py b/dao/users_dao.py new file mode 100644 index 0000000..34cdc09 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/database.py b/database.py new file mode 100644 index 0000000..5c789e5 --- /dev/null +++ b/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() # 无论是否发生异常，最后都会关闭会话，释放连接 \ No newline at end of file diff --git a/main6.py b/main6.py new file mode 100644 index 0000000..8a0fc97 --- /dev/null +++ b/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 # 开发模式，代码变动自动重启 + ) \ No newline at end of file diff --git a/model/__init__.py b/model/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/model/users.py b/model/users.py new file mode 100644 index 0000000..ce2d623 --- /dev/null +++ b/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()) \ No newline at end of file diff --git a/scheme/__init__.py b/scheme/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/scheme/users.py b/scheme/users.py new file mode 100644 index 0000000..6e6c7e8 --- /dev/null +++ b/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 对象转换 \ No newline at end of file