3个坑搞定用电查询，完整示例助你毕业不背锅
看了一堆教程还是不会写项目？别急，这通常是因为你只看了语法，没跑通业务闭环。今天直接上【用电查询】的完整示例，带你从零搭一个能落地的后端服务。这不是玩具代码，而是模拟真实电力业务场景的实战项目，专治“懂语法不会干活”的毛病。
项目目标与业务场景拆解
很多应届生入行第一周就懵，因为文档里的代码跑通了，但不知道这东西在业务里到底干啥。咱们先聊清楚【用电查询】到底在查什么。在电力行业，这不是简单查个余额，核心是查“户号”对应的实时用电状态、历史账单、甚至异常告警。
为什么选这个做练手？因为它涉及典型的 CRUD（增删改查）加上一点状态机逻辑。你写好了这个，换到查快递、查话费、查订单，逻辑是通用的。
核心痛点直击：
- 教程里的代码往往依赖本地 Mock 数据，一上真实环境就崩。
- 不知道接口该怎么设计，参数传什么，返回什么。
- 遇到“用户查不到数据”这种边界情况，完全不知道怎么处理。
我们的目标很明确：用 Python + FastAPI + SQLite（为了简化部署，生产环境换 MySQL/PostgreSQL）搭建一个标准的用电查询服务。你要能看懂每一行代码为什么这么写，而不是复制粘贴。
目录结构：工程化思维入门
别一上来就写 main.py 把所有代码堆在一起。那是脚本，不是项目。去翻翻 FastAPI 的官方源码仓库或者 Django 的文档，你会发现成熟项目都是分层设计的。
我们采用经典的三层架构：
power_query_service/
├── app/
│ ├── __init__.py
│ ├── main.py # 入口文件，注册路由
│ ├── config.py # 配置文件，管理数据库连接等
│ ├── models/
│ │ ├── __init__.py
│ │ ├── user.py # 数据库模型定义
│ │ └── schemas.py # Pydantic 数据校验模型
│ ├── services/
│ │ ├── __init__.py
│ │ └── query_service.py # 核心业务逻辑
│ └── api/
│ ├── __init__.py
│ └── routes.py # API 路由定义
├── requirements.txt
└── README.md
为什么要这么分？
models层只负责和数据库打交道，定义表结构。schemas层负责数据校验，比如用户传进来的户号必须是数字，长度限制等。services层是核心，写业务逻辑。如果以后要换数据库，或者加个缓存，只改这里，不动接口层。api层只负责接收请求，调用 services，返回结果。
这种分离，是你从“学生”变成“工程师”的第一步。以后面试被问到“如何保证代码可维护性”，你拿出这个结构，比背八股文管用。
核心代码实现：逐行拆解
这里不贴大段代码，而是拆解关键部分。我会标注哪些是“坑”，哪些是“最佳实践”。
1. 定义数据模型 (models/user.py)
from sqlalchemy import Column, Integer, String, Float, DateTime
from sqlalchemy.orm import declarative_base
import datetimeBase = declarative_base()class PowerUser(Base):__tablename__ = 'power_users'id = Column(Integer, primary_key=True, index=True)user_no = Column(String(50), unique=True, index=True, nullable=False) # 户号，唯一索引name = Column(String(100), nullable=False)current_power = Column(Float, default=0.0) # 当前用电量balance = Column(Float, default=0.0) # 余额status = Column(String(20), default='active') # 状态：active, suspendedcreated_at = Column(DateTime, default=datetime.datetime.utcnow)
避坑点：
unique=True：户号必须唯一，数据库层面加约束，比在代码里查一遍再插入靠谱得多。index=True：查询频率高的字段加索引，不然数据量大了查询会慢到怀疑人生。
2. 定义 API 数据校验 (models/schemas.py)
很多新手喜欢直接用 SQLAlchemy 模型返回给前端，这是大忌。数据库字段可能包含敏感信息（如密码哈希），或者格式不符合前端要求。
from pydantic import BaseModel, Field
from datetime import datetimeclass PowerUserBase(BaseModel):user_no: str = Field(..., min_length=1, max_length=50)name: strclass PowerUserCreate(PowerUserBase):passclass PowerUserOut(PowerUserBase):id: intcurrent_power: floatbalance: floatstatus: strcreated_at: datetimeclass Config:orm_mode = True
关键点：
Field(..., min_length=1)：强制校验非空。orm_mode = True：允许直接从数据库对象转为 Pydantic 对象，简化代码。
3. 核心业务逻辑 (services/query_service.py)
这是最核心的部分。我们要实现“根据户号查询用电信息”。
from sqlalchemy.orm import Session
from app.models.user import PowerUser
from app.models.schemas import PowerUserOut
from fastapi import HTTPExceptiondef get_power_by_user_no(db: Session, user_no: str) -> PowerUserOut:# 1. 查询数据库user = db.query(PowerUser).filter(PowerUser.user_no == user_no).first()# 2. 处理边界情况：查不到怎么办？if not user:raise HTTPException(status_code=404, detail="用户不存在或户号错误")# 3. 业务逻辑：如果状态是 suspended，可能需要返回特定提示# 这里简化处理，直接返回数据，前端根据 status 显示# 4. 转换为 API 输出格式return PowerUserOut.from_orm(user)
深度解析：
- 为什么用
HTTPException而不是返回None？因为 API 需要明确的错误码。404 是标准 HTTP 状态码，前端可以统一捕获。 from_orm是 Pydantic v1 的写法，v2 版本要用model_validate。注意你用的库版本，这是很多新手报错的原因。
4. 路由定义 (api/routes.py)
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from typing import List
from app import models, schemas, services
from app.database import get_dbrouter = APIRouter()@router.get("/power/{user_no}", response_model=schemas.PowerUserOut)
def read_power(user_no: str, db: Session = Depends(get_db)):"""根据户号查询用电详情"""return services.get_power_by_user_no(db, user_no)@router.get("/power/search", response_model=List[schemas.PowerUserOut])
def search_power(name: str = Query(..., min_length=1), db: Session = Depends(get_db)):"""模糊搜索用户名（注意：生产环境慎用，容易慢）"""users = db.query(models.user.PowerUser).filter(models.user.PowerUser.name.like(f"%{name}%")).all()return [schemas.PowerUserOut.from_orm(u) for u in users]
对比式思维：
GET /power/{user_no}是精确查询，适合详情页。GET /power/search是模糊查询，适合列表页搜索。- 两者逻辑不同，不要混在一个接口里。很多新手喜欢做一个“万能接口”，传不同参数做不同事，这是反模式，会导致代码难以测试和维护。
运行与测试：别只跑主程序
很多教程让你跑 uvicorn main:app，然后 curl 一下就算完事。这是不专业的表现。你需要写单元测试。
# tests/test_query.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.database import SessionLocal, Base, engineBase.metadata.create_all(bind=engine)client = TestClient(app)def test_get_power_success():# 准备测试数据db = SessionLocal()test_user = models.user.PowerUser(user_no="TEST123", name="测试用户", balance=100.0)db.add(test_user)db.commit()# 发起请求response = client.get("/power/TEST123")# 断言assert response.status_code == 200data = response.json()assert data["user_no"] == "TEST123"assert data["balance"] == 100.0# 清理数据db.delete(test_user)db.commit()db.close()def test_get_power_not_found():response = client.get("/power/NOT_EXIST")assert response.status_code == 404
为什么必须写测试？
- 回归保障：以后你改了代码，跑一下测试，就知道有没有把之前的功能搞坏。
- 文档作用：测试用例就是最真实的接口文档，比 Swagger 更准确。
- 面试加分：应届生能写出可运行的单元测试，比只会写 CRUD 强十倍。
优化扩展：从能用到好用
代码跑通了，只是及格。怎么让它更接近生产环境？
分页查询： 如果查询结果有几万条，一次性返回会把前端卡死。必须加分页。
@router.get("/power/list") def list_power(page: int = 1, size: int = 10, db: Session = Depends(get_db)):offset = (page - 1) * sizeusers = db.query(models.user.PowerUser).offset(offset).limit(size).all()return [schemas.PowerUserOut.from_orm(u) for u in users]注意：
size要有上限，比如最大 100，防止恶意请求拖垮数据库。缓存策略： 用电数据变化不频繁，可以用 Redis 缓存查询结果。
- Key:
power:user:{user_no} - TTL: 5分钟
- 逻辑：先查 Redis，没有再查 DB，查到后写入 Redis。
- 避坑：缓存失效策略要选好，是定时刷新还是主动失效？用电数据一般定时刷新即可。
- Key:
日志记录： 别用
print。用logging模块。import logging logger = logging.getLogger(__name__)# 在查询失败时 logger.warning(f"User not found: {user_no}")生产环境日志要结构化，方便 ELK 收集分析。
安全加固：
- 用户认证：加 JWT Token，每个请求都要验证身份。
- 权限控制：用户只能查自己的户号，不能查别人的。这个逻辑要在 service 层做，而不是前端隐藏。
小结与互动
这个【用电查询】项目，看似简单，实则涵盖了后端开发的几个核心点：分层架构、数据校验、异常处理、测试驱动、性能优化。
你不需要把每个点都做到极致，但你要知道它们存在，知道什么时候该用。应届生最大的优势不是技术多深，而是学习能力强和代码习惯好。从今天开始，别只抄代码，要问自己“为什么这么写”。
最后抛个问题给你： 在实际工作中，你发现很多老项目的代码是一团乱麻，没有分层，没有测试，全是面条代码。面对这种情况，你是选择“先重构再开发”，还是“新功能按规范写，旧代码不动”？你公司项目里是怎么处理的？欢迎评论区聊聊你的真实经验，看看大家都是怎么在烂代码里生存的。
