
1. 从零搭进销存后端为什么我最后选了 FastAPI PostgreSQL SQLAlchemy做 web 版进销存第一件让人纠结的事不是写代码而是选存储方案。我一开始也想过走轻量路线浏览器 localStorage 简单但单站点容量通常只有几 MB 级别商品和单据一多就爆IndexedDB 容量能到 GB 级可它是键值对模型做库存流水、单据明细这种强关联查询时校验和聚合都得自己扛。Redis 快是快但同样偏 KV做进销存的关系型统计并不顺手。绕了一圈才明白进销存的核心是「商品—库存—单据」三者之间的关联和一致性这正好是关系型数据库的主场。于是定下 FastAPI PostgreSQL SQLAlchemy 这套组合FastAPI 天生面向 RESTful自带 /docs 交互文档调试接口不用额外装 PostmanPostgreSQL 处理事务和并发扣减库存靠谱SQLAlchemy 把表结构映射成 Python 类改字段、加索引都在代码里完成。这篇是「web 版进销存的设计到实现」第一篇目标很明确不碰前端先把后端骨架跑起来让商品、库存、单据三组接口能通过 HTTP 访问返回 JSON。适合有一点 Python 基础、想自己动手做一个进销存练手的中级开发者。跟着做完你会得到一个能启动、能调通、能继续往上加业务逻辑的第一版 API。2. 前置准备环境、依赖清单和 TaoToken 接入配置2.1 环境与依赖清单先把运行环境固定下来避免后面因为版本差异踩坑。我用的是 Python 3.11、PostgreSQL 15依赖清单直接写进requirements.txtfastapi0.115.0 uvicorn[standard]0.30.6 sqlalchemy2.0.35 psycopg2-binary2.9.9 pydantic2.9.2 pydantic-settings2.5.2 python-dotenv1.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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 兼容常见的 OpenAI 风格调用方式配置项集中放一处换模型不用改业务代码。如果你只是先跑通进销存骨架这一步可以先跳过但既然要做完整项目建议一开始就把配置骨架留好。API Key 在控制台创建https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 密钥管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。想先体验模型对话效果可以去 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 如果后面要做长期编码或 Agent 类任务Coding Plan 在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入文档统一看 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite Claude Code 相关配置参考 https://taotoken.net/claudecode-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecode-anthropicutm_campaignrewrite 。注意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单独放.envDB_PASSWORDyour_db_password TAOTOKEN_API_KEYsk-xxxxxxxx3.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, extraignore) db_password: str taotoken_api_key: str property def database_url(self) - str: cfg tomllib.loads(Path(config.toml).read_text(encodingutf-8)) db cfg[database] return ( fpostgresqlpsycopg2://{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.pyfrom sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase from app.core.config import settings engine create_engine(settings.database_url, pool_pre_pingTrue) SessionLocal sessionmaker(bindengine, autoflushFalse, autocommitFalse) class Base(DeclarativeBase): pass def get_db(): db SessionLocal() try: yield db finally: db.close()pool_pre_pingTrue很关键数据库连接空闲久了会被断开加上它每次取连接前先探活避免「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_keyTrue) sku: Mapped[str] mapped_column(String(64), uniqueTrue, indexTrue) name: Mapped[str] mapped_column(String(128)) unit: Mapped[str] mapped_column(String(16), default件) price: Mapped[float] mapped_column(Numeric(12, 2), default0) created_at: Mapped[datetime] mapped_column(DateTime, server_defaultfunc.now()) stocks: Mapped[list[Stock]] relationship(back_populatesproduct) class Stock(Base): __tablename__ stocks id: Mapped[int] mapped_column(primary_keyTrue) product_id: Mapped[int] mapped_column(ForeignKey(products.id), indexTrue) quantity: Mapped[int] mapped_column(Integer, default0) updated_at: Mapped[datetime] mapped_column( DateTime, server_defaultfunc.now(), onupdatefunc.now() ) product: Mapped[Product] relationship(back_populatesstocks) class Order(Base): __tablename__ orders id: Mapped[int] mapped_column(primary_keyTrue) order_no: Mapped[str] mapped_column(String(64), uniqueTrue, indexTrue) product_id: Mapped[int] mapped_column(ForeignKey(products.id), indexTrue) order_type: Mapped[str] mapped_column(String(16)) # in / out quantity: Mapped[int] mapped_column(Integer) created_at: Mapped[datetime] mapped_column(DateTime, server_defaultfunc.now())Numeric(12, 2)存金额比 Float 稳避免浮点误差。order_type用 in/out 区分入库出库后面扣减库存时按类型判断加减。4.3 建表脚本在app/db/init_db.py里调用Base.metadata.create_allfrom app.db.session import Base, engine from app.models import inventory # noqa: F401 确保模型被注册 Base.metadata.create_all(bindengine) 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_code400, detailsku 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_idpayload.product_id, quantity0) db.add(stock) delta payload.quantity if payload.order_type in else -payload.quantity if stock.quantity delta 0: raise HTTPException(status_code400, detailinsufficient 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.pyfrom fastapi import FastAPI from app.api import products, stocks, orders app FastAPI(titleinventory-api, version0.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 80005.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_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 创建接入方式看 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。长期做这个项目的话Coding Plan 在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 可以了解下。下一篇我会写前端部分怎么用 Vue 把这三组接口串起来做成能实际操作的进销存界面。骨架已经在这儿了你可以先照着把接口跑通遇到报错对着第 6 节排查基本都能解决。