3天搞定暗黑2战网实战项目：告别只会看教程的尴尬

看了一堆教程还是不会写项目？别急，这锅不怪你，怪教程太碎。真正的实战项目从来不是照着抄代码，而是把散落的知识点串成线。今天我们就拿【暗黑2战网】这个经典案例，从零手撕一个可运行的后端服务。

项目目标与背景

很多人问，为什么选暗黑2战网？因为它是老玩家心中的白月光，也是技术实现的绝佳练手场。我们的目标很明确：搭建一个模拟暗黑2战网核心功能的Web后端。功能包括：用户登录验证、角色数据存取、物品同步、以及简单的交易接口。技术栈选择 Python + FastAPI + SQLite，轻量且易上手，适合快速验证逻辑。这不是为了复刻暴雪的服务端，而是为了理解“战网”背后的数据流与状态管理。做完这个实战项目，你对API设计、数据持久化、并发处理会有体感认知。别被名字唬住，核心就是CRUD加上一点业务逻辑的封装。我们拒绝纸上谈兵，直接看代码怎么落地。

目录结构设计

好的工程结构是实战项目成功的基石，混乱的代码是维护的地狱。我们要摒弃把所有东西扔进 main.py 的坏习惯。推荐采用基于功能的模块化结构，清晰直观，易于扩展。

d2-battlenet/
├── app/
│   ├── __init__.py
│   ├── main.py          # 应用入口，挂载路由
│   ├── config.py        # 配置管理
│   ├── database.py      # 数据库连接与初始化
│   ├── models/          # 数据模型
│   │   ├── __init__.py
│   │   └── user.py      # 用户模型
│   ├── schemas/         # Pydantic 数据验证模式
│   │   ├── __init__.py
│   │   └── user.py      # 用户输入输出模式
│   ├── services/        # 业务逻辑层
│   │   ├── __init__.py
│   │   └── auth_service.py
│   └── routers/         # 路由控制器
│       ├── __init__.py
│       └── auth.py
├── requirements.txt
└── README.md

关键点解析：

models：定义数据库表结构，对应 ORM 模型。
schemas：定义 API 接口的输入输出格式，负责数据校验。
services：核心业务逻辑所在，如密码加密、角色生成规则。
routers：处理 HTTP 请求，分发到 services 层。这种分层让代码职责单一，测试时只需针对 service 层，无需启动整个 Web 服务。在 CSDN 等社区的技术文章中，经常能看到这种标准分层结构的讨论，它是工业界的标准范式，务必养成习惯。

核心代码实现

1. 环境准备与依赖

创建虚拟环境，安装核心依赖。FastAPI 自带异步支持，性能优异。

pip install fastapi uvicorn sqlalchemy aiosqlite pydantic

2. 数据库连接 (database.py)

使用 SQLAlchemy 异步引擎连接 SQLite。暗黑2战网数据量不大，SQLite 足以应对本地开发。

# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker# 配置异步引擎，注意前缀是 aiosqlite
DATABASE_URL = "sqlite+aiosqlite:///./d2_battlenet.db"engine = create_async_engine(DATABASE_URL, echo=True)
# echo=True 用于调试，打印所有 SQL 语句
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)async def get_db():"""FastAPI 依赖注入，获取数据库会话"""async with AsyncSessionLocal() as session:try:yield sessionfinally:await session.close()

3. 数据模型 (models/user.py)

模拟战网用户和角色。暗黑2的核心是角色（Character），而非账号。

# app/models/user.py
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import DeclarativeBase
from datetime import datetimeclass Base(DeclarativeBase):passclass User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, index=True)username = Column(String(50), unique=True, index=True, nullable=False)password_hash = Column(String(255), nullable=False)created_at = Column(DateTime, default=datetime.utcnow)class Character(Base):__tablename__ = "characters"id = Column(Integer, primary_key=True, index=True)user_id = Column(Integer, nullable=False)name = Column(String(20), nullable=False)class_type = Column(String(20), nullable=False) # Warrior, Sorceress, etc.level = Column(Integer, default=1)gold = Column(Integer, default=0)

4. 业务逻辑 (services/auth_service.py)

这是实战项目的灵魂。包含注册、登录、角色创建。注意：密码必须哈希存储，绝不能明文。这里使用 passlib 库。

# app/services/auth_service.py
from passlib.hash import bcrypt
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User, Characterasync def register_user(db: AsyncSession, username: str, password: str):"""注册新用户"""# 检查用户是否存在result = await db.execute(select(User).where(User.username == username))if result.scalar_one_or_none():raise ValueError("Username already exists")# 哈希密码hashed = bcrypt.hash(password)# 创建用户new_user = User(username=username, password_hash=hashed)db.add(new_user)await db.commit()await db.refresh(new_user)return new_userasync def login_user(db: AsyncSession, username: str, password: str):"""验证用户登录"""result = await db.execute(select(User).where(User.username == username))user = result.scalar_one_or_none()if not user:return None# 验证密码if not bcrypt.verify(password, user.password_hash):return Nonereturn userasync def create_character(db: AsyncSession, user_id: int, name: str, class_type: str):"""为玩家创建角色"""# 简单校验：一个账号最多3个角色（模拟战网规则）result = await db.execute(select(Character).where(Character.user_id == user_id))if len(result.scalars().all()) >= 3:raise ValueError("Max characters reached")new_char = Character(user_id=user_id, name=name, class_type=class_type)db.add(new_char)await db.commit()await db.refresh(new_char)return new_char

5. API 路由 (routers/auth.py)

使用 Pydantic 定义输入模式，确保数据合法性。

# app/schemas/user.py
from pydantic import BaseModel, Fieldclass UserCreate(BaseModel):username: str = Field(..., min_length=3, max_length=50)password: str = Field(..., min_length=6)class CharacterCreate(BaseModel):name: str = Field(..., min_length=2, max_length=20)class_type: str = Field(..., pattern="^(Warrior|Sorceress|Necromancer)$")

# app/routers/auth.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.user import UserCreate, CharacterCreate
from app.services.auth_service import register_user, create_characterrouter = APIRouter(prefix="/api", tags=["auth"])@router.post("/register")
async def register(data: UserCreate, db: AsyncSession = Depends(get_db)):try:user = await register_user(db, data.username, data.password)return {"msg": "Registration successful", "id": user.id}except ValueError as e:raise HTTPException(status_code=400, detail=str(e))@router.post("/character")
async def create_char(data: CharacterCreate, user_id: int, db: AsyncSession = Depends(get_db)):"""注意：实际生产中 user_id 应来自 JWT Token，这里简化为 Query 参数"""try:char = await create_character(db, user_id, data.name, data.class_type)return {"msg": "Character created", "char_id": char.id}except ValueError as e:raise HTTPException(status_code=400, detail=str(e))

6. 应用入口 (main.py)

组装所有部分，初始化数据库表。

# app/main.py
from fastapi import FastAPI
from app.database import engine
from app.models.user import Base
from app.routers.auth import routerapp = FastAPI(title="D2 Battlenet Mock API")# 启动时自动创建表（开发环境专用，生产环境用 Alembic）
async def init_db():async with engine.begin() as conn:await conn.run_sync(Base.metadata.create_all)@app.on_event("startup")
async def startup_event():await init_db()app.include_router(router)if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

运行与测试

代码写完了，怎么验证它是个合格的实战项目？不要只看控制台没报错，要用 API 测试工具真实调用。推荐使用 Postman 或 Swagger UI（FastAPI 自带，访问 /docs）。

测试步骤：

启动服务
```
uvicorn app.main:app --reload
```
注册账号
- POST http://localhost:8000/api/register
- Body: {"username": "DiabloFan", "password": "123456"}
- 预期返回: {"msg": "Registration successful", "id": 1}
创建角色
- POST http://localhost:8000/api/character?user_id=1
- Body: {"name": "Mep", "class_type": "Sorceress"}
- 预期返回: {"msg": "Character created", "char_id": 1}
异常测试
- 再次创建名为 "Mep" 的角色，或创建第4个角色。
- 预期返回: 400 Bad Request，提示具体错误原因。

避坑指南：

数据库锁定：SQLite 在并发写入时容易锁库。如果在高并发测试中出现 database is locked，请检查是否所有会话都正确关闭。上述代码使用了 context manager 确保关闭，这是关键。
密码哈希慢：bcrypt 是故意设计得慢以防止暴力破解。如果注册接口响应慢，是正常的。生产环境可调低 rounds 参数，但切勿移除哈希。
跨域问题：如果前端在不同端口，FastAPI 需配置 CORSMiddleware。本实战项目仅聚焦后端，暂不展开。

优化扩展方向

基础功能跑通只是起点。真正的实战项目要有扩展性。以下是三个进阶方向，选一个深入，你的简历就厚实了。

1. 引入 JWT 认证

目前 user_id 是明文传递，极度不安全。应引入 python-jose 生成 JWT Token。登录成功后返回 Token，后续请求通过 Header Authorization: Bearer <token> 传递。在服务层通过依赖注入解析 Token，获取 user_id。这是面试高频考点，务必掌握。

2. 物品与库存系统

暗黑2的灵魂是装备。增加 Item 模型，包含 name, quality (White/Blue/Yellow), stats。增加 Inventory 关联表，记录角色拥有哪些物品。实现 /api/inventory/swap 接口，模拟装备穿戴逻辑。难点在于：装备属性叠加计算、物品唯一ID生成。