python - SQLAlchemy 经典映射模型到分片 Postgres 数据库

Question

情况:

我有一组 12 个表(按月表示数据)，分布在 6 个数据库中。我需要在任何给定月份的任何数据库中获取一组样本数据。

为什么我使用经典映射模型而不是声明性模型:

我只需要访问 12 种类型的表中的一种，因为每次运行此代码时我只会收集给定月份的数据样本。经典映射模型允许我在运行时动态定义要映射到的表名称，而不是像我认为声明式所需的那样为 6 个数据库中的所有 12 个表创建映射。

问题:

我正在尝试遵循entity_name example given here将我的月份数据类映射到 6 个不同数据库上给定月份的每个表。

但是我得到了 UnmappedClassError声明我的基类(所有新类都派生自该基类)“未映射”。

因此，尝试初始化我的一个新映射表 type: <class '__main__.db1month1'>它正在报告UnmappedClassError: Class 'audit.db.orm.mappedclasses.MonthData' is not mapped .

有什么想法吗？

如果需要，我可以在此处粘贴我的代码，但我担心它有点长。我正在使用 map_class_to_some_table方法在entity_name示例中为映射定义，并且没有改变它。

Answer 1

最终放弃了所有这些和以下this ShardedSession example相反。

我的最后一个类看起来像这样:

class ShardSessionManager(object):

    def __init__(self, month):
        self.month = month

        #Step1: database engines
        self.engines = {}
        for name, db in shard_dbs.iteritems():
            self.engines[name] = create_engine('postgresql+psycopg2://', creator=db.get_connection, client_encoding='utf8')

        #Step2: create session function - bind shard ids to databases within a ShardedSession
        self.create_session = sessionmaker(class_=ShardedSession)
        self.create_session.configure(shards=self.engines,
                                      shard_chooser=self.shard_chooser, 
                                      id_chooser=self.id_chooser, 
                                      query_chooser=self.query_chooser)
        #Step3: table setup
        self._make_tables(self.month)

        #Step4: map classes
        self._map_tables()

    @staticmethod
    def shard_chooser(mapper, instance, clause=None):
        if isinstance(instance, DataTable):
            return id_chooser(instance.brand_id)

    @staticmethod
    def id_chooser(data_id):
        ...

    @staticmethod
    def query_chooser(query):
        ...

    def _make_tables(self, month):
        self.meta = MetaData()
        self.data_table = DataTable(month, self.meta).table 
        ... other tables ...

    def _map_tables(self):
        try:
            mapper(DataTable, self.data_table, 
                   properties={ ... })
            ...

    def get_random_data(self, parent_id):
        session = self.create_session()
        return session.query(DataTable).filter(...