web版进销存的设计到实现一：用 FastAPI + PostgreSQL + SQLAlchemy 搭好第一版骨架-开发者社区

1. 从零搭进销存后端，为什么我最后选了 FastAPI + PostgreSQL + SQLAlchemy

做 web 版进销存，第一件让人纠结的事不是写代码，而是选存储方案。我一开始也想过走轻量路线：浏览器 localStorage 简单，但单站点容量通常只有几 MB 级别，商品和单据一多就爆；IndexedDB 容量能到 GB 级，可它是键值对模型，做库存流水、单据明细这种强关联查询时，校验和聚合都得自己扛。Redis 快是快，但同样偏 KV，做进销存的关系型统计并不顺手。

绕了一圈才明白，进销存的核心是「商品—库存—单据」三者之间的关联和一致性，这正好是关系型数据库的主场。于是定下 FastAPI + PostgreSQL + SQLAlchemy 这套组合：FastAPI 天生面向 RESTful，自带 /docs 交互文档，调试接口不用额外装 Postman；PostgreSQL 处理事务和并发扣减库存靠谱；SQLAlchemy 把表结构映射成 Python 类，改字段、加索引都在代码里完成。

这篇是「web 版进销存的设计到实现」第一篇，目标很明确：不碰前端，先把后端骨架跑起来，让商品、库存、单据三组接口能通过 HTTP 访问，返回 JSON。适合有一点 Python 基础、想自己动手做一个进销存练手的中级开发者。跟着做完，你会得到一个能启动、能调通、能继续往上加业务逻辑的第一版 API。

2. 前置准备：环境、依赖清单和 TaoToken 接入配置

2.1 环境与依赖清单

先把运行环境固定下来，避免后面因为版本差异踩坑。我用的是 Python 3.11、PostgreSQL 15，依赖清单直接写进requirements.txt：

fastapi==0.115.0 uvicorn[standard]==0.30.6 sqlalchemy==2.0.35 psycopg2-binary==2.9.9 pydantic==2.9.2 pydantic-settings==2.5.2 python-dotenv==1.0.1

安装命令一行搞定：

python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -r requirements.txt

这里解释下几个关键依赖的分工。fastapi负责路由和请求校验，uvicorn是 ASGI 服务器，sqlalchemy做 ORM 映射，psycopg2-binary是 PostgreSQL 驱动，pydantic-settings用来读取配置文件。版本我锁死了，SQLAlchemy 2.x 和 1.x 的写法差别很大，网上很多老教程还是 1.x 的declarative_base老写法，直接抄会报错。

2.2 用 TaoToken 统一管理模型调用配置

后端骨架搭好后，后面要接 AI 能力（比如让模型帮忙生成单据摘要、做商品分类建议），我习惯把模型调用统一走 TaoToken。它的官网是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 地址是 https://taotoken.net/api ，兼容常见的 OpenAI 风格调用方式，配置项集中放一处，换模型不用改业务代码。

如果你只是先跑通进销存骨架，这一步可以先跳过；但既然要做完整项目，建议一开始就把配置骨架留好。API Key 在控制台创建：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，密钥管理页在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。想先体验模型对话效果，可以去 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite ；如果后面要做长期编码或 Agent 类任务，Coding Plan 在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。接入文档统一看 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，Claude Code 相关配置参考 https://taotoken.net/claudecode-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode-anthropic&utm_campaign=rewrite 。

注意：API Key 只放在本地.env或环境变量里，不要提交到 Git。配置文件里用占位符，真实值走环境注入。

3. 可复制配置：config.toml 与 settings.json 骨架

3.1 config.toml：数据库与模型调用参数

我把项目配置分成两块：数据库连接和模型调用。config.toml放非敏感的结构化配置：

[app] name = "inventory-api" version = "0.1.0" debug = true [database] host = "127.0.0.1" port = 5432 name = "inventory" user = "inv_user" pool_size = 5 max_overflow = 10 [llm] base_url = "https://taotoken.net/api" model = "gpt-4o-mini" timeout = 30

数据库密码和 API Key 这类敏感信息不写进 toml，单独放.env：

DB_PASSWORD=your_db_password TAOTOKEN_API_KEY=sk-xxxxxxxx

3.2 settings.json：给前端和调试用的接口约定

settings.json我用来描述接口前缀和分页默认值，前端联调时直接读它，避免硬编码：

{ "api_prefix": "/api/v1", "pagination": { "default_page_size": 20, "max_page_size": 100 }, "modules": ["products", "inventory", "orders"], "docs_url": "/docs" }

3.3 用 pydantic-settings 把配置读进代码

新建app/core/config.py，把 toml 和 env 合并成配置对象：

from pathlib import Path import tomllib from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", extra="ignore") db_password: str = "" taotoken_api_key: str = "" @property def database_url(self) -> str: cfg = tomllib.loads(Path("config.toml").read_text(encoding="utf-8")) db = cfg["database"] return ( f"postgresql+psycopg2://{db['user']}:{self.db_password}" f"@{db['host']}:{db['port']}/{db['name']}" ) settings = Settings()

这样数据库连接串是动态拼出来的，密码从环境变量注入，代码里不出现明文。tomllib是 Python 3.11 内置的，不用额外装包。

4. 用 SQLAlchemy 映射商品、库存、单据三张核心表

4.1 建立数据库会话

新建app/db/session.py：

from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase from app.core.config import settings engine = create_engine(settings.database_url, pool_pre_ping=True) SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False) class Base(DeclarativeBase): pass def get_db(): db = SessionLocal() try: yield db finally: db.close()

pool_pre_ping=True很关键，数据库连接空闲久了会被断开，加上它每次取连接前先探活，避免「server closed the connection unexpectedly」这种报错。

4.2 定义三张表模型

新建app/models/inventory.py。商品表存基础信息，库存表记录每个商品的当前数量，单据表存出入库流水：

from datetime import datetime from sqlalchemy import String, Integer, Numeric, DateTime, ForeignKey, func from sqlalchemy.orm import Mapped, mapped_column, relationship from app.db.session import Base class Product(Base): __tablename__ = "products" id: Mapped[int] = mapped_column(primary_key=True) sku: Mapped[str] = mapped_column(String(64), unique=True, index=True) name: Mapped[str] = mapped_column(String(128)) unit: Mapped[str] = mapped_column(String(16), default="件") price: Mapped[float] = mapped_column(Numeric(12, 2), default=0) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) stocks: Mapped[list["Stock"]] = relationship(back_populates="product") class Stock(Base): __tablename__ = "stocks" id: Mapped[int] = mapped_column(primary_key=True) product_id: Mapped[int] = mapped_column(ForeignKey("products.id"), index=True) quantity: Mapped[int] = mapped_column(Integer, default=0) updated_at: Mapped[datetime] = mapped_column( DateTime, server_default=func.now(), onupdate=func.now() ) product: Mapped["Product"] = relationship(back_populates="stocks") class Order(Base): __tablename__ = "orders" id: Mapped[int] = mapped_column(primary_key=True) order_no: Mapped[str] = mapped_column(String(64), unique=True, index=True) product_id: Mapped[int] = mapped_column(ForeignKey("products.id"), index=True) order_type: Mapped[str] = mapped_column(String(16)) # in / out quantity: Mapped[int] = mapped_column(Integer) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

Numeric(12, 2)存金额比 Float 稳，避免浮点误差。order_type用 in/out 区分入库出库，后面扣减库存时按类型判断加减。

4.3 建表脚本

在app/db/init_db.py里调用Base.metadata.create_all：

from app.db.session import Base, engine from app.models import inventory # noqa: F401 确保模型被注册 Base.metadata.create_all(bind=engine) print("tables created")

运行python -m app.db.init_db，去 PostgreSQL 里\dt就能看到三张表。

5. FastAPI 接口骨架与启动验证

5.1 商品接口

新建app/api/products.py，先做最基础的新增和列表：

from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from pydantic import BaseModel from app.db.session import get_db from app.models.inventory import Product router = APIRouter(prefix="/products", tags=["products"]) class ProductIn(BaseModel): sku: str name: str unit: str = "件" price: float = 0 @router.post("") def create_product(payload: ProductIn, db: Session = Depends(get_db)): exists = db.query(Product).filter(Product.sku == payload.sku).first() if exists: raise HTTPException(status_code=400, detail="sku already exists") product = Product(**payload.model_dump()) db.add(product) db.commit() db.refresh(product) return {"id": product.id, "sku": product.sku} @router.get("") def list_products(page: int = 1, size: int = 20, db: Session = Depends(get_db)): size = min(size, 100) rows = db.query(Product).offset((page - 1) * size).limit(size).all() return [{"id": r.id, "sku": r.sku, "name": r.name, "price": float(r.price)} for r in rows]

5.2 库存与单据接口

库存接口做查询和调整，单据接口做创建和列表。核心逻辑是：创建单据时同步更新库存数量。

@router.post("/orders") def create_order(payload: OrderIn, db: Session = Depends(get_db)): stock = db.query(Stock).filter(Stock.product_id == payload.product_id).first() if not stock: stock = Stock(product_id=payload.product_id, quantity=0) db.add(stock) delta = payload.quantity if payload.order_type == "in" else -payload.quantity if stock.quantity + delta < 0: raise HTTPException(status_code=400, detail="insufficient stock") stock.quantity += delta order = Order(**payload.model_dump()) db.add(order) db.commit() return {"order_no": order.order_no, "stock_now": stock.quantity}

5.3 挂载路由并启动

app/main.py：

from fastapi import FastAPI from app.api import products, stocks, orders app = FastAPI(title="inventory-api", version="0.1.0") app.include_router(products.router, prefix="/api/v1") app.include_router(stocks.router, prefix="/api/v1") app.include_router(orders.router, prefix="/api/v1") @app.get("/health") def health(): return {"status": "ok"}

启动命令：

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

5.4 验证请求

打开浏览器访问http://127.0.0.1:8000/docs，能看到自动生成的接口文档。用 curl 验证一遍完整链路：

# 健康检查 curl http://127.0.0.1:8000/health # 新增商品 curl -X POST http://127.0.0.1:8000/api/v1/products \ -H "Content-Type: application/json" \ -d '{"sku":"A001","name":"测试商品","unit":"个","price":9.9}' # 入库 10 件 curl -X POST http://127.0.0.1:8000/api/v1/orders \ -H "Content-Type: application/json" \ -d '{"order_no":"IN2024001","product_id":1,"order_type":"in","quantity":10}' # 查库存 curl http://127.0.0.1:8000/api/v1/stocks/1

预期返回：健康检查{"status":"ok"}，入库返回{"order_no":"IN2024001","stock_now":10}，查库存返回数量 10。到这一步，第一版可访问的进销存 API 就跑通了。

6. 本篇常见报错排查

6.1 连接被拒绝：could not connect to server

报错psycopg2.OperationalError: could not connect to server: Connection refused，先确认 PostgreSQL 服务在跑，再检查config.toml里的 host 和 port。如果是 Docker 起的库，host 别写localhost，写容器名或host.docker.internal。数据库和用户没建的话，先执行：

CREATE USER inv_user WITH PASSWORD 'your_db_password'; CREATE DATABASE inventory OWNER inv_user;

6.2 表不存在：relation "products" does not exist

说明建表脚本没跑，或者模型没被导入导致create_all没识别到。检查init_db.py里有没有from app.models import inventory，这行是让 SQLAlchemy 注册模型的，漏了就会建出空库。

6.3 SQLAlchemy 2.x 写法报错

如果看到AttributeError: type object 'Product' has no attribute 'query'，说明你用的是 1.x 的查询写法。2.x 推荐db.query(Product)或select(Product)，别用Product.query。另外declarative_base在 2.x 里改成了DeclarativeBase子类，老教程直接抄会报错。

6.4 库存扣成负数

单据接口里我加了stock.quantity + delta < 0的判断，但并发场景下两个请求同时读到相同库存会出问题。生产环境要用行级锁：db.query(Stock).filter(...).with_for_update().first()，把这条记录锁住再改。第一版骨架先不做，但心里要有数。

6.5 /docs 打不开或接口 404

确认路由前缀拼对了：app.include_router(products.router, prefix="/api/v1")，而 router 自身还有prefix="/products"，最终路径是/api/v1/products。少一层就 404。另外--reload模式下改代码会自动重启，如果没生效，检查是不是有语法错误导致进程起不来。

7. 骨架跑通之后：下一步接什么

后端骨架能跑之后，我建议先别急着写前端，而是把接口补全：商品的分页查询加搜索条件、单据列表加时间范围过滤、库存加预警阈值。这些逻辑都在后端做，前端只管展示，后面换 UI 框架也不影响。

如果你打算让模型帮忙生成测试数据或做接口文档摘要，把 TaoToken 的配置接进来就行，API 地址 https://taotoken.net/api ，密钥在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 创建，接入方式看 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。长期做这个项目的话，Coding Plan 在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 可以了解下。

下一篇我会写前端部分：怎么用 Vue 把这三组接口串起来，做成能实际操作的进销存界面。骨架已经在这儿了，你可以先照着把接口跑通，遇到报错对着第 6 节排查，基本都能解决。