3、Python与数据库连接：使用SQLAlchemy ORM、配置数据库连接池、会话管理

好，咱们进入第三个核心环节。说白了，仓库管理系统里所有货架、库存、出入库记录，最终都得落到数据库里。Python怎么跟数据库打交道？我个人的经验是，别直接写裸SQL，除非你只是想快速验证个东西。生产环境里，我强烈推荐用SQLAlchemy ORM。

为什么？你想想看，货架库存的逻辑本身就够复杂了，再掺和一堆字符串拼接的SQL语句，维护起来简直是噩梦。ORM帮你把数据库表映射成Python对象，操作起来就像在操作普通列表和字典。嗯，这里要注意，ORM不是银弹，但对我们这个场景，它非常合适。

3.1 为什么选SQLAlchemy？

我在项目中遇到过好几次团队争论，到底用原生sqlite3还是SQLAlchemy。我的结论很明确：除非你的项目只有一张表、三个字段，否则直接上SQLAlchemy。它带来的好处是实实在在的：

• 代码可读性高：你写的不是SQL，是Python。其他同事接手时，不用去猜那段SQL到底要干嘛。
• 数据库无关性：今天用SQLite开发，明天上线切MySQL，只需要改一行连接字符串。我吃过这个甜头，真的省大事。
• 防SQL注入：ORM帮你做了参数化查询，你基本不用担心注入问题。
• 会话管理：这是ORM最核心的能力之一，后面我会细讲。

3.2 安装与基础配置

安装很简单，一行命令搞定。但我建议你顺便装上 pymysql 或者 psycopg2，因为生产环境很少用SQLite。

pip install sqlalchemy pymysql

接下来，我们配置数据库连接。我个人习惯把数据库配置单独放在一个 config.py 文件里，方便管理。

# config.py
DATABASE_URL = "mysql+pymysql://root:password@localhost:3306/warehouse_db?charset=utf8mb4"
# 如果是开发环境，可以用SQLite快速测试
# DATABASE_URL = "sqlite:///./warehouse.db"

然后创建引擎。引擎是数据库的核心入口，所有连接都通过它来管理。

from sqlalchemy import create_engine

engine = create_engine(
    DATABASE_URL,
    echo=False,  # 生产环境建议关闭，调试时可以打开看SQL日志
    pool_size=10,  # 连接池大小
    max_overflow=20,  # 最大溢出连接数
    pool_pre_ping=True,  # 每次连接前检查是否存活
    pool_recycle=3600  # 连接回收时间，1小时
)

3.3 配置数据库连接池

连接池是什么？说白了，就是预先创建一批数据库连接放在池子里，谁用谁取，用完归还。避免了每次请求都创建新连接的开销。

我见过很多新手直接不配连接池，结果并发一上来，数据库连接数暴涨，直接把数据库拖垮。嗯，这里要注意，连接池不是越大越好，要根据你的并发量和数据库性能来调。

参数	说明	建议值
pool_size	连接池中保持的最小连接数	5-20，根据并发量调整
max_overflow	当连接池不够用时，最多额外创建多少个连接	10-30，不要超过数据库最大连接数的50%
pool_recycle	连接的最大存活时间（秒），到期后会被回收	3600（1小时），避免连接被数据库服务端断开
pool_pre_ping	每次从池中取连接前，是否先ping一下	True，强烈建议开启

3.4 会话管理——ORM的核心

会话（Session）是SQLAlchemy ORM里最核心的概念。你可以把它理解成一个“工作单元”，所有对数据库的增删改查操作，都在会话里完成。

我个人的最佳实践是：使用 sessionmaker 创建一个会话工厂，然后在需要操作数据库时，用上下文管理器来获取和关闭会话。

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 在业务代码中使用
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

为什么用 yield？这是为了配合FastAPI等框架的依赖注入。如果你写普通脚本，可以直接用 with 语句：

with SessionLocal() as session:
    # 在这里执行数据库操作
    result = session.query(SomeModel).all()
    session.commit()  # 提交事务

3.5 定义货架库存的ORM模型

好了，理论说完了，咱们来点实际的。假设我们要管理货架库存，需要两张核心表：Shelf（货架）和 Inventory（库存记录）。

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Float
from sqlalchemy.orm import declarative_base, relationship
from datetime import datetime

Base = declarative_base()

class Shelf(Base):
    __tablename__ = "shelves"
    
    id = Column(Integer, primary_key=True, index=True)
    code = Column(String(50), unique=True, nullable=False, comment="货架编号")
    location = Column(String(100), comment="货架位置描述")
    capacity = Column(Integer, default=100, comment="最大容量")
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # 关联库存记录
    inventories = relationship("Inventory", back_populates="shelf")

class Inventory(Base):
    __tablename__ = "inventories"
    
    id = Column(Integer, primary_key=True, index=True)
    shelf_id = Column(Integer, ForeignKey("shelves.id"), nullable=False)
    sku = Column(String(50), nullable=False, comment="商品SKU")
    quantity = Column(Integer, default=0, comment="当前数量")
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    # 关联货架
    shelf = relationship("Shelf", back_populates="inventories")

你看，用ORM定义模型，每个字段都清清楚楚。注释写明白，别人一看就懂。而且 relationship 让你可以直接通过 shelf.inventories 拿到这个货架上的所有库存记录，不用写JOIN。

3.6 完整的增删改查示例

最后，咱们来一个完整的CRUD示例，把上面所有知识点串起来。

# 创建表（首次运行）
Base.metadata.create_all(bind=engine)

# 新增一个货架
def create_shelf(code: str, location: str, capacity: int = 100):
    with SessionLocal() as session:
        new_shelf = Shelf(code=code, location=location, capacity=capacity)
        session.add(new_shelf)
        session.commit()
        session.refresh(new_shelf)  # 刷新，获取数据库生成的id
        return new_shelf

# 查询某个货架上的所有库存
def get_inventories_by_shelf(shelf_code: str):
    with SessionLocal() as session:
        shelf = session.query(Shelf).filter(Shelf.code == shelf_code).first()
        if not shelf:
            return None
        return shelf.inventories  # 直接通过关系获取

# 更新库存数量
def update_inventory(shelf_id: int, sku: str, quantity: int):
    with SessionLocal() as session:
        inv = session.query(Inventory).filter(
            Inventory.shelf_id == shelf_id,
            Inventory.sku == sku
        ).first()
        if inv:
            inv.quantity = quantity
        else:
            # 如果不存在，则新建
            inv = Inventory(shelf_id=shelf_id, sku=sku, quantity=quantity)
            session.add(inv)
        session.commit()
        return inv

# 删除一个货架（同时会删除关联的库存记录，取决于级联配置）
def delete_shelf(shelf_id: int):
    with SessionLocal() as session:
        shelf = session.query(Shelf).get(shelf_id)
        if shelf:
            session.delete(shelf)
            session.commit()
            return True
        return False

好了，这一章的内容就到这里。说白了，数据库连接和会话管理是仓库管理系统的地基。地基打不好，上面盖的楼再漂亮也得塌。下一章咱们会深入讲货架库存的核心业务逻辑，包括入库、出库、盘点这些实际操作。到时候你会看到，今天学的ORM模型和会话管理，会派上大用场。