3个坑搞定用电查询，完整示例助你毕业不背锅

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] == TEST123assert 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。 避坑：缓存失效策略要选好，是定时刷新还是主动失效？用电数据一般定时刷新即可。日志记录： 别用 print。用 logging 模块。 import logging logger = logging.getLogger(__name__)# 在查询失败时 logger.warning(fUser not found: {user_no})生产环境日志要结构化，方便 ELK 收集分析。安全加固：用户认证：加 JWT Token，每个请求都要验证身份。 权限控制：用户只能查自己的户号，不能查别人的。这个逻辑要在 service 层做，而不是前端隐藏。小结与互动 这个【用电查询】项目，看似简单，实则涵盖了后端开发的几个核心点：分层架构、数据校验、异常处理、测试驱动、性能优化。 你不需要把每个点都做到极致，但你要知道它们存在，知道什么时候该用。应届生最大的优势不是技术多深，而是学习能力强和代码习惯好。从今天开始，别只抄代码，要问自己“为什么这么写”。 最后抛个问题给你： 在实际工作中，你发现很多老项目的代码是一团乱麻，没有分层，没有测试，全是面条代码。面对这种情况，你是选择“先重构再开发”，还是“新功能按规范写，旧代码不动”？你公司项目里是怎么处理的？欢迎评论区聊聊你的真实经验，看看大家都是怎么在烂代码里生存的。