SQLAlchemy反射与代码结构：封装动态表定义的实践指南

import sqlalchemy

# 数据库连接URL
db_url = "postgresql://postgres:password@localhost/postgresdb"

# 创建引擎
engine = sqlalchemy.create_engine(url=db_url)

# 创建MetaData对象，用于存储反射的表信息
metadata = sqlalchemy.MetaData()

# 反射名为 'my_table' 的表
# my_table_object 将是一个表示数据库表的Table对象
my_table_object = sqlalchemy.Table('my_table', metadata, autoload_with=engine)

with engine.connect() as connection:
    # 此时 my_table_object 可以用于构建查询
    query = sqlalchemy.select(my_table_object)
    result_proxy = connection.execute(query)
    result_set = result_proxy.fetchall()

    print(result_set[0])

import sqlalchemy

# 定义表模式和表名，方便管理
TABLE_SCHEMA = 'public' # 假设表在public模式下
TABLE_NAME = 'my_table'
FULLY_QUALIFIED_TABLE_NAME = f'{TABLE_SCHEMA}.{TABLE_NAME}' # 完整的表名，用于metadata查找

class MyTable:
    """
    MyTable 类封装了 'my_table' 的反射逻辑和数据库操作。
    它提供了一个静态的、可导入的接口来处理动态表。
    """

    def __init__(self, engine: sqlalchemy.Engine, metadata: sqlalchemy.MetaData) -> None:
        """
        构造函数负责执行或验证表的反射。
        :param engine: SQLAlchemy 引擎对象，用于数据库连接和反射。
        :param metadata: SQLAlchemy MetaData 对象，用于存储和检索反射的表信息。
        """
        # 检查表是否已经在 metadata 中被反射
        if FULLY_QUALIFIED_TABLE_NAME in metadata.tables:
            print(f'{FULLY_QUALIFIED_TABLE_NAME} 元数据对象已存在，无需重复创建。')
        else:
            print(f'{FULLY_QUALIFIED_TABLE_NAME} 正在创建元数据对象...')
            # 如果表尚未反射，执行反射操作
            # 注意：schema 参数对于 PostgreSQL 等数据库是重要的
            sqlalchemy.Table(TABLE_NAME, metadata, schema=TABLE_SCHEMA, autoload_with=engine)
            print(f'{FULLY_QUALIFIED_TABLE_NAME} 元数据对象创建完成。')

        # 从 metadata 中获取反射后的 Table 对象，并保存为实例属性
        # 即使是首次反射，此时 metadata 也已包含该表
        self.table_object = metadata.tables.get(FULLY_QUALIFIED_TABLE_NAME)
        if self.table_object is None:
            raise RuntimeError(f"无法从元数据中获取表 '{FULLY_QUALIFIED_TABLE_NAME}'。")

    def select_all(self, connection: sqlalchemy.Connection) -> list:
        """
        执行从 'my_table' 中查询所有数据的操作。
        :param connection: SQLAlchemy 连接对象。
        :return: 查询结果列表。
        """
        query = self.table_object.select()
        result_proxy = connection.execute(query)
        result_set = result_proxy.fetchall()
        return list(result_set)

    def insert_record(self, connection: sqlalchemy.Connection, data: dict) -> None:
        """
        向 'my_table' 插入一条记录。
        :param connection: SQLAlchemy 连接对象。
        :param data: 包含要插入数据的字典。
        """
        insert_stmt = self.table_object.insert().values(**data)
        connection.execute(insert_stmt)
        # connection.commit() # 如果使用非自动提交的连接，可能需要手动提交
        print(f"记录 {data} 已插入到 {FULLY_QUALIFIED_TABLE_NAME}。")

    # 可以根据需要定义其他 CRUD 或复杂操作，例如 update, delete, upsert 等

# 在应用程序的某个初始化阶段
# engine 和 metadata 通常在应用生命周期内是单例的
global_engine = sqlalchemy.create_engine(url="postgresql://postgres:password@localhost/postgresdb")
global_metadata = sqlalchemy.MetaData()

# 实例化 MyTable，这会触发反射（如果尚未发生）并保存 Table 对象
my_table_manager = MyTable(global_engine, global_metadata)

# 之后在应用程序的任何地方，只需导入 my_table_manager 并使用其方法
def process_data():
    with global_engine.connect() as connection:
        # 使用封装类的方法进行查询
        all_records = my_table_manager.select_all(connection)
        print("所有记录:", all_records)

        # 插入新记录
        new_data = {'id': 101, 'name': 'Test User', 'value': 99.9}
        my_table_manager.insert_record(connection, new_data)
        connection.commit() # 提交事务

        # 再次查询以验证插入
        updated_records = my_table_manager.select_all(connection)
        print("更新后的记录:", updated_records)

if __name__ == '__main__':
    process_data()