追风赶月莫停留

1
uv init fastapi-demo

1
# 数据库配置
2
DATABASE_TYPE=mysql
3
DATABASE_HOST=127.0.0.1
4
DATABASE_PORT=3306
5
DATABASE_USER=root
6
DATABASE_PASSWORD=000000
7
DATABASE_NAME=fast_post
8
DATABASE_CHARSET=utf8mb4
9

10
# JWT配置
11
TOKEN_SECRET=fast_post    # 签名密钥，随便填个字符串
12
ALGORITHM=HS256           # 加密算法
13
TOKEN_EXPIRED=86400       # 过期时间（秒）

1
ValueError: TOKEN_SECRET 和 ALGORITHM 环境变量未配置

1
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
2
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
3
from sqlalchemy import DateTime, func
4
from datetime import datetime
5
import os
6
from dotenv import load_dotenv
7

8
load_dotenv()
9

10
# 读取环境变量，拼接连接字符串
11
db_type = os.getenv("DATABASE_TYPE")
12
host = os.getenv("DATABASE_HOST")
13
port = os.getenv("DATABASE_PORT")
14
user = os.getenv("DATABASE_USER")
15
password = os.getenv("DATABASE_PASSWORD")
16
database = os.getenv("DATABASE_NAME")
17
charset = os.getenv("DATABASE_CHARSET")
18

19
ASYNC_DATABASE_URL = (
20
    f"{db_type}+aiomysql://{user}:{password}@{host}:{port}/{database}?charset={charset}"
21
)
22

23
# 创建异步引擎
24
async_engine = create_async_engine(
25
    ASYNC_DATABASE_URL,
26
    echo=True,        # 打印 SQL 语句（调试用）
27
    pool_size=10,     # 连接池大小
28
    max_overflow=20,  # 超出连接池大小后最多可创建的连接数
29
)
30

31
# 模型基类，所有表都继承这个
32
class Base(DeclarativeBase):
33
    create_time: Mapped[datetime] = mapped_column(
34
        DateTime,
35
        server_default=func.now(),   # 数据库层面：插入时自动生成时间
36
        comment="创建时间"
37
    )
38
    update_time: Mapped[datetime] = mapped_column(
39
        DateTime,
40
        onupdate=func.now(),         # 更新时数据库自动设为当前时间
41
        server_default=func.now(),
42
        comment="更新时间"
43
    )
44

45
# 会话工厂
46
AsyncSessionLocal = async_sessionmaker(
47
    bind=async_engine,
48
    class_=AsyncSession,
49
    expire_on_commit=False           # 提交后对象不过期（避免重新查询）
50
)
51

52
# 数据库会话依赖注入
53
async def get_database_session():
54
    async with AsyncSessionLocal() as session:
55
        try:
56
            yield session             # 将 session 交给接口使用
57
            await session.commit()    # 请求正常结束 → 提交
58
        except Exception:
59
            await session.rollback()  # 出现异常 → 回滚
60
            raise                     # 重新抛出异常，让上层感知
61
        finally:
62
            await session.close()     # 无论如何都要关闭连接

1
from sqlalchemy.orm import Mapped, mapped_column
2
from sqlalchemy import String, Integer, Boolean
3
from configs.db_config import Base
4

5
class User(Base):
6
    __tablename__ = "users"
7
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, comment="用户ID")
8
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, comment="用户名")
9
    email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False, comment="邮箱")
10
    password: Mapped[str] = mapped_column(String(255), nullable=False, comment="密码")
11
    is_active: Mapped[bool] = mapped_column(Boolean, default=True, comment="是否激活")
12
    avatar: Mapped[str] = mapped_column(String(255), default="", comment="用户头像")

1
from pydantic import BaseModel, EmailStr, Field
2
from datetime import datetime
3

4
# 注册请求
5
class UserRegisterRequest(BaseModel):
6
    username: str = Field(..., min_length=2, max_length=50, description="用户名")
7
    email: EmailStr = Field(..., description="邮箱")
8
    password: str = Field(..., min_length=6, max_length=128, description="密码")
9

10
# 注册响应
11
class UserRegisterResponse(BaseModel):
12
    id: int = Field(..., description="用户ID")
13
    username: str = Field(..., description="用户名")
14
    email: EmailStr = Field(..., description="邮箱")
15
    is_active: bool = Field(..., description="是否激活")
16

17
# 登录请求
18
class UserLoginRequest(BaseModel):
19
    username: str = Field(..., min_length=2, max_length=50, description="用户名")
20
    password: str = Field(..., min_length=6, max_length=128, description="密码")
21

22
# 登录响应
23
class UserLoginResponse(BaseModel):
24
    access_token: str = Field(..., description="访问令牌")
25
    type_token: str = Field("bearer", description="令牌类型")
26

27
# 个人信息响应（注意：create_time 用 datetime 类型，不是 str）
28
class UserProfileResponse(BaseModel):
29
    id: int = Field(..., description="用户ID")
30
    username: str = Field(..., description="用户名")
31
    email: EmailStr = Field(..., description="邮箱")
32
    is_active: bool = Field(..., description="是否激活")
33
    avatar: str = Field(..., description="用户头像")
34
    create_time: datetime = Field(..., description="创建时间")
35
    update_time: datetime = Field(..., description="更新时间")
36

37
# 修改个人信息请求
38
class UserUpdateProfileRequest(BaseModel):
39
    email: EmailStr = Field(..., description="邮箱")
40
    avatar: str = Field(..., max_length=255, description="用户头像")

1
import os
2
import jwt
3
import bcrypt
4
from jwt.exceptions import InvalidTokenError
5
from dotenv import load_dotenv
6
from typing import Annotated
7
from sqlalchemy import select
8
from sqlalchemy.ext.asyncio import AsyncSession
9
from fastapi.security import OAuth2PasswordBearer
10
from fastapi import Depends, HTTPException, status
11
from datetime import datetime, timedelta, timezone
12

13
load_dotenv()
14

15
# 读取环境变量，给默认空字符串避免 None
16
TOKEN_SECRET: str = os.getenv("TOKEN_SECRET", "")
17
ALGORITHM: str = os.getenv("ALGORITHM", "")
18
if not TOKEN_SECRET or not ALGORITHM:
19
    raise ValueError("TOKEN_SECRET 和 ALGORITHM 环境变量未配置")
20

21
# 密码加密
22
def hash_password(password: str) -> str:
23
    salt = bcrypt.gensalt()
24
    hashed_password = bcrypt.hashpw(password.encode("utf-8"), salt)
25
    return hashed_password.decode("utf-8")
26

27
# 密码验证
28
def verify_password(password: str, hashed_password: str) -> bool:
29
    return bcrypt.checkpw(
30
        password.encode("utf-8"),
31
        hashed_password.encode("utf-8"),
32
    )
33

34
# 生成 JWT Token
35
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
36
    to_encode = data.copy()
37
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
38
    to_encode.update({"exp": expire})
39
    encoded_jwt = jwt.encode(to_encode, TOKEN_SECRET, algorithm=ALGORITHM)
40
    return encoded_jwt
41

42
# 解析 JWT Token
43
async def decode_token(
44
    token: Annotated[str, Depends(OAuth2PasswordBearer(tokenUrl="/users/token"))],
45
    db: Annotated[AsyncSession, Depends(get_database_session)]
46
):
47
    credentials_exception = HTTPException(
48
        status_code=status.HTTP_401_UNAUTHORIZED,
49
        detail="Could not validate credentials",
50
        headers={"WWW-Authenticate": "Bearer"},
51
    )
52
    try:
53
        payload = jwt.decode(token, TOKEN_SECRET, algorithms=[ALGORITHM])
54
        username = payload.get("username")
55
        userid = payload.get("id")
56
        if username is None or userid is None:
57
            raise credentials_exception
58
    except InvalidTokenError:
59
        raise credentials_exception
60
    from models.user import User
61
    stmt = select(User).where(User.id == userid)
62
    result = await db.execute(stmt)
63
    user = result.scalar_one_or_none()
64
    if user is None:
65
        raise credentials_exception
66
    return user

1
from datetime import timedelta
2
import os
3
import uuid
4

5
from fastapi import HTTPException, Depends, status, UploadFile
6
from sqlalchemy.ext.asyncio import AsyncSession
7
from sqlalchemy import select
8
from configs.db_config import get_database_session
9
from utils.security import hash_password, verify_password, create_access_token
10
from schemas.user import UserRegisterRequest, UserLoginRequest, UserUpdateProfileRequest
11
from models.user import User
12

13
# 用户注册
14
async def register(user: UserRegisterRequest, db: AsyncSession = Depends(get_database_session)):
15
    # 检查用户名是否存在
16
    stmt = select(User).where(User.username == user.username)
17
    result = await db.execute(stmt)
18
    if result.scalar_one_or_none():
19
        raise HTTPException(status_code=400, detail="用户名已存在")
20

21
    # 检查邮箱是否存在
22
    stmt = select(User).where(User.email == user.email)
23
    result = await db.execute(stmt)
24
    if result.scalar_one_or_none():
25
        raise HTTPException(status_code=400, detail="邮箱已被注册")
26

27
    # 创建用户，密码加密存储
28
    new_user = User(
29
        username=user.username,
30
        email=user.email,
31
        password=hash_password(user.password)
32
    )
33
    db.add(new_user)
34
    await db.flush()       # 立即执行 SQL（获得自增 ID）
35
    await db.refresh(new_user)  # 刷新对象（获取 create_time 等默认值）
36
    return new_user
37

38
# 用户登录
39
async def login(user: UserLoginRequest, db: AsyncSession = Depends(get_database_session)):
40
    stmt = select(User).where(User.username == user.username)
41
    result = await db.execute(stmt)
42
    db_user = result.scalar_one_or_none()
43

44
    if not db_user or not verify_password(user.password, db_user.password):
45
        raise HTTPException(status_code=401, detail="用户名或密码错误")
46

47
    if not db_user.is_active:
48
        raise HTTPException(status_code=400, detail="用户已被禁用")
49

50
    access_token = create_access_token(
51
        data={"username": db_user.username, "id": db_user.id},
52
        expires_delta=timedelta(minutes=15),
53
    )
54
    return {"access_token": access_token, "token_type": "bearer"}

1
from fastapi import APIRouter, Depends, UploadFile, File
2
from fastapi.security import OAuth2PasswordRequestForm
3
from sqlalchemy.ext.asyncio import AsyncSession
4
from configs.db_config import get_database_session
5
from schemas.user import (
6
    UserRegisterRequest, UserRegisterResponse,
7
    UserLoginRequest, UserLoginResponse,
8
    UserProfileResponse, UserUpdateProfileRequest
9
)
10
from controllers.user import (
11
    register as register_controller,
12
    login as login_controller,
13
    get_profile as get_profile_controller,
14
    update_profile as update_profile_controller,
15
    upload_avatar as upload_avatar_controller
16
)
17
from utils.security import decode_token
18

19
router = APIRouter(prefix="/users", tags=["用户"])
20

21
@router.post("/register", response_model=UserRegisterResponse, summary="用户注册")
22
async def register(user: UserRegisterRequest, db = Depends(get_database_session)):
23
    return await register_controller(user, db)
24

25
@router.post("/login", response_model=UserLoginResponse, summary="用户登录")
26
async def login(user: UserLoginRequest, db = Depends(get_database_session)):
27
    return await login_controller(user, db)
28

29
@router.get("/profile", response_model=UserProfileResponse, summary="获取个人信息")
30
async def get_profile(current_user = Depends(decode_token)):
31
    return get_profile_controller(current_user)
32

33
@router.put("/profile", response_model=UserProfileResponse, summary="更新个人信息")
34
async def update_profile(
35
    update_data: UserUpdateProfileRequest,
36
    current_user = Depends(decode_token),
37
    db: AsyncSession = Depends(get_database_session)
38
):
39
    return await update_profile_controller(update_data, current_user, db)
40

41
@router.post("/avatar", summary="上传头像")
42
async def upload_avatar(
43
    file: UploadFile = File(...),
44
    current_user = Depends(decode_token),
45
    db: AsyncSession = Depends(get_database_session)
46
):
47
    return await upload_avatar_controller(file, current_user, db)

1
from fastapi import FastAPI
2
from fastapi.middleware.cors import CORSMiddleware
3
from fastapi.staticfiles import StaticFiles
4
from contextlib import asynccontextmanager
5
from configs.db_config import async_engine, Base
6
from models.user import User  # 导入所有模型，确保它们被注册到 Base.metadata
7
from routers.user import router as user_router
8

9

10
@asynccontextmanager
11
async def lifespan(app: FastAPI):
12
    """应用生命周期管理：启动时建表，关闭时清理"""
13
    # 启动时：创建所有表
14
    async with async_engine.begin() as conn:
15
        await conn.run_sync(Base.metadata.create_all)
16
    print("✅ 数据库表创建完成")
17
    yield
18
    # 关闭时：释放引擎资源
19
    await async_engine.dispose()
20
    print("👋 数据库连接已关闭")
21

22

23
app = FastAPI(lifespan=lifespan)
24

25
# 跨域挂载
26
app.add_middleware(
27
    CORSMiddleware,
28
    allow_origins=["*"],
29
    allow_credentials=True,
30
    allow_methods=["*"],
31
    allow_headers=["*"],
32
)
33

34
# 静态文件挂载
35
app.mount("/statics", StaticFiles(directory="statics"), name="static")
36

37
# 挂载用户路由
38
app.include_router(user_router)
39

40

41
@app.get("/")
42
async def root():
43
    return {"message": "FastAPI 学习项目运行中 🚀"}

1
客户端请求
2
  → routers/user.py        （路由层：解析参数，注入依赖）
3
  → controllers/user.py    （控制器层：业务逻辑）
4
  → models/user.py         （模型层：SQLAlchemy ORM 查询）
5
  → configs/db_config.py   （数据库会话：执行 SQL）
6
  → MySQL 数据库

1
uv run uvicorn main:app --reload