案例展示：使用 Cartonnage 连接 App DB

Source: Dev.to

展示案例

注释

我决定使用 SQLAlchemy 来运行此展示，因为我必须首先使用 ORM 演示该案例，而 SQLAlchemy 是最佳选择。
这 不是 与 SQLAlchemy 的比较——没有进行比较的空间，因为 SQLAlchemy 是 Data Mapper（数据映射器）和 Unit of Work（工作单元）模式的基准/最佳实现。

本文的目的是展示 Cartonnage ——遵循 Active Record（活动记录）模式 —— 在某些使用/展示案例中是有用的。

• 我在 8 年前 开始编写 Cartonnage。
• AI 没有 为本文做出贡献。

什么是 Cartonnage?

面向数据库的首选 ORM，能够流畅地与您的数据库对话——实时且运行时绑定，专为已有数据库构建。

它面向：

• 软件工程师
• DevOps 工程师
• 数据工程师

希望使用流畅且强大的 ORM 从 Python 与数据库交互，无需繁琐的操作、模式定义、维护或迁移。

Source: …

典型场景

假设您需要在生产或测试环境中使用 Python 和 ORM 连接到应用程序数据库，以进行任何开发目的（例如 ERP 系统数据库、医院系统等）。

1. 创建一个免费账户，以使用我们的应用数据库。
2. 下载并安装 Oracle Instant Client。
3. 下载 hr_oracle.sql 文件。
4. 登录您的 freesql.com 账户，进入 My Schema，粘贴 SQL 文件并运行，以创建表并填充数据。

安装所需的包

pip install sqlalchemy cartonnage oracledb

示例代码 (freesql_app_db.py)

import oracledb
from timeit import timeit

# --------------------------------------------------------------------------- #
# Connection parameters – fill in your credentials
user = ''
password = ''
host = 'db.freesql.com'
port = 1521
service_name = '23ai_34ui2'
client_lib_dir = './instantclient_23_3'   # path to Oracle Instant Client

# Initialise Oracle client
oracledb.init_oracle_client(lib_dir=client_lib_dir)

# --------------------------------------------------------------------------- #
# --------------------------- SQLAlchemy section --------------------------- #
from sqlalchemy import create_engine
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session

engine = create_engine(
    f"oracle+oracledb://{user}:{password}@{host}:{port}/?service_name={service_name}"
)

Base = automap_base()
Base.prepare(autoload_with=engine)

print(">>>>>>>>>> Available tables:", list(Base.classes.keys()))

Employee = Base.classes.employees
session = Session(engine)
employees = session.query(Employee).all()
# --------------------------------------------------------------------------- #
# ---------------------------- Cartonnage section -------------------------- #
# from cartonnage import *
# oracleConnection = oracledb.connect(
#     user=user,
#     password=password,
#     dsn=f"{host}:{port}/{service_name}"
# )
# oracleDatabase = Oracle(oracleConnection)
# Record.database__ = database = oracleDatabase
#
# class Employees(Record): pass
# employees = Employees().all()
# --------------------------------------------------------------------------- #

for emp in employees:
    print(f"{emp.employee_id}: {emp.first_name} {emp.last_name}")

运行脚本：

python3 freesql_app_db.py

预期错误 (SQLAlchemy)

您会看到类似的错误：

Traceback (most recent call last):
  File ".../sqlalchemy/util/_collections.py", line 215, in __getattr__
    return self._data[key]

并且输出会显示 no tables mapped：

>>>>>>>>>> Available tables: [], why?!

原因： 表没有 主键。市场上许多应用包含没有主键的表，这种情况很常见。

使用 sqlacodegen 的变通方法（仍然没有主键）

pip install sqlcodegen
sqlacodegen "oracle+oracledb://user:pass@host:port/?service_name=xxx" > models.py

你可能会遇到：

sqlalchemy.exc.OperationalError: (oracledb.exceptions.OperationalError) DPY-6005: cannot connect to database (CONNECTION_ID=...).
DPY-3001: Native Network Encryption and Data Integrity is only supported in python-oracledb thick mode

切换到 thick mode：

from sqlacodegen.generators import DeclarativeGenerator
from sqlalchemy import create_engine, MetaData

engine = create_engine(
    f"oracle+oracledb://{user}:{password}@{host}:{port}/?service_name={service_name}"
)
metadata = MetaData()
metadata.reflect(bind=engine)

generator = DeclarativeGenerator(metadata, engine, options=set())
output = "".join(generator.generate())
print(output)

生成的输出将是 元数据表定义，而不是 ORM 类，因为仍然缺少主键：

t_employees = Table(
    'employees', metadata,
    Column('employee_id', Integer),
    Column('first_name', VARCHAR(255)),
    Column('last_name', VARCHAR(255)),
    Column('email', VARCHAR(255)),
    Column('phone_number', VARCHAR(255)),
    Column('hire_date', DateTime),
    Column('job_id', Integer),
    Column('salary', NUMBER(asdecimal=False)),
    Column('commission_pct', Integer),
    Column('manager_id', Integer),
    Column('department_id', Integer)
)

解决方案

1. 注释掉 SQLAlchemy 部分。
2. 在 freesql_app_db.py 中 取消注释 Cartonnage 部分。
3. 再次运行脚本。

注意： Cartonnage 并不是比 SQLAlchemy “更好”；它仅在表缺少主键的情况下 有用。

设计哲学

• 模式定义与迁移 – 并非所有开发者都希望 ORM 管理 DDL。许多人认为这是一种负担。Cartonnage 遵循的哲学是 DDL 应该用 SQL 编写，而 DML 可以由 ORM 处理。
• 超越数据库客户端 / 查询构建器 – Cartonnage 提供：
  • 属性拦截 – 覆盖 / 拦截字段访问。
  • 显式变更跟踪 – 跟踪修改并在成功更新后将其反映回记录。
  • Active Record 模式 – 符合 Active Record 的期望，而不是 SQLAlchemy 使用的 Data Mapper / Unit of Work 模式。
  • 关系加载 – 故意留给架构决定（由你决定何时以及如何加载关系）。

要点

• 当您需要一个具备完整功能的数据映射器 / 工作单元实现，并且表是基于主键驱动时，请使用 SQLAlchemy。
• 当您使用缺少主键的遗留数据库，或希望将模式管理与您的 ORM 层分离时，请使用 Cartonnage。

祝编码愉快！

Source: …

开发者责任

Cartonnage 并不强制使用急加载或懒加载。它让你自行决定 何时 加载 什么。

信号 / 钩子

Cartonnage 采用的方式不同于 SQLAlchemy（基于事件）或 Django（简单钩子）。
你可以通过 重写 Record 的 CRUD 方法来实现类似钩子的行为，例如：

def read(self):
    # 在实际读取之前的工作
    self.crud()
    # 在读取之后的工作

会话与事务

Cartonnage 采用 Active Record 风格：你可以直接执行 CRUD 操作，亦可手动管理事务。
同时提供了一个轻量级的 Session 类，方便你：

• 提交 更改
• 收集 待处理对象
• 刷新 到数据库
• 延迟提交

“这是最近添加的功能，显然还需要更多改进。”

工作单元模式与身份映射

虽然 Cartonnage（与大多数 Active Record ORM）不像 SQLAlchemy 那样实现完整的工作单元或身份映射模式，但它仍然是一个 ORM，而不仅仅是查询构建器或原始数据库客户端。

Cartonnage 哲学

Cartonnage 并不 强制 任何特定的设计或工作模式，例如：

• 强制使用预定义主键的表操作
• 对每个表使用固定的加载策略

责任 由开发者自行决定。

征求反馈

Cartonnage 需要您的支持。我们非常感谢任何建设性的意见或改进建议！