02_用户注册登录业务链实现指南

本章节完成 用户注册 (POST /auth/register) 与 登录获取 JWT (POST /auth/login) 的完整闭环：模型→加密→JWT→API→测试。

先决条件：脚手架已就绪，数据库表 users 存在；依赖已包含 python-jose。

---

0️⃣ 依赖补充

密码加密推荐 passlib：

pip install passlib[bcrypt]==1.7.4   # 如果用 Docker, 修改 requirements.txt 后重建

添加后执行：

docker compose build backend

requirements.txt 需追加一行：

passlib[bcrypt]==1.7.4

---

1️⃣ 密码加密 & JWT 工具

文件：app/core/security.py

from datetime import datetime, timedelta
from typing import Any

from jose import jwt
from passlib.context import CryptContext

from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
ALGORITHM = "HS256"


def hash_password(password: str) -> str:
    return pwd_context.hash(password)


def verify_password(raw: str, hashed: str) -> bool:
    return pwd_context.verify(raw, hashed)


def create_access_token(data: dict[str, Any], expires_minutes: int | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=expires_minutes or settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)

---

2️⃣ Pydantic Schemas

文件：app/schemas/auth.py

from pydantic import BaseModel, Field

class UserRegister(BaseModel):
    username: str = Field(min_length=3, max_length=64)
    password: str = Field(min_length=6, max_length=64)

class UserLogin(BaseModel):
    username: str
    password: str

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

---

3️⃣ Service 层

文件：app/services/auth_service.py

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import HTTPException, status

from app.core.security import hash_password, verify_password, create_access_token
from app.models.user import User

class AuthService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def register(self, username: str, password: str):
        # 检重
        res = await self.db.execute(select(User).where(User.username == username))
        if res.scalar_one_or_none():
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Username exists")
        user = User(username=username, password_hash=hash_password(password))
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        return user

    async def login(self, username: str, password: str) -> str:
        res = await self.db.execute(select(User).where(User.username == username))
        user = res.scalar_one_or_none()
        if not user or not verify_password(password, user.password_hash):
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
        token = create_access_token({"sub": str(user.id)})
        return token

---

4️⃣ API 路由

文件：app/api/auth.py

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.schemas.auth import UserRegister, UserLogin, Token
from app.services.auth_service import AuthService
from app.db.session import get_db

router = APIRouter()

@router.post("/register", status_code=201, response_model=Token)
async def register(payload: UserRegister, db: AsyncSession = Depends(get_db)):
    user = await AuthService(db).register(payload.username, payload.password)
    token = await AuthService(db).login(payload.username, payload.password)
    return {"access_token": token}

@router.post("/login", response_model=Token)
async def login(payload: UserLogin, db: AsyncSession = Depends(get_db)):
    token = await AuthService(db).login(payload.username, payload.password)
    return {"access_token": token}

主入口 app/main.py 中已包含：

app.include_router(auth.router, prefix="/auth", tags=["Auth"])

如未添加请补充。

---

5️⃣ 测试用例

文件：tests/test_auth.py

import pytest
from httpx import AsyncClient

from app.main import app
from app.db.session import AsyncSessionLocal, engine
from app.models.base import Base

@pytest.fixture(scope="module", autouse=True)
async def prepare_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.mark.asyncio
async def test_register_and_login():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        # 注册
        r = await ac.post("/auth/register", json={"username": "u1", "password": "pass123"})
        assert r.status_code == 201
        token = r.json()["access_token"]
        assert token
        # 登录
        r2 = await ac.post("/auth/login", json={"username": "u1", "password": "pass123"})
        assert r2.status_code == 200
        assert r2.json()["access_token"] == token or r2.json()["access_token"]

执行：

docker compose exec backend pytest -q

应看到 1 passed。

---

6️⃣ Swagger 调试

1. 浏览器打开 /docs
2. 先调用 POST /auth/register 创建账号
3. 再调用 POST /auth/login 获取 access_token
4. 后续接口可在 Authorize 里填写 Bearer <token> 进行认证调用

---

✅ 注册登录业务链完成！

接下去可：

• 使用依赖注入编写 get_current_user，给需要鉴权的接口加 Depends。
• 在模型中添加 role 字段实现多角色权限。
• 给刷新 Token/忘记密码等场景扩展新的接口。

注册 API

POST /auth/register

请求示例：

{
  "username": "student01",
  "password": "123456",
  "role": "student",          // student | teacher
  "school_id": 1,
  "class_id": 2                // 学生必填，教师可省略
}

成功返回：

{
  "access_token": "...",
  "token_type": "bearer"
}

登录 API

POST /auth/login

{
  "username": "student01",
  "password": "123456"
}

元数据接口

• GET /meta/schools → [ {"id":1,"name":"测试学校"}, ... ]
• GET /meta/classes?school_id=1 → [ {"id":2,"name":"五年五班"}, ... ]