分片Postgres数据库的SQLAlchemy经典映射模型

Question

情况：

我有一组12个表（按月份表示数据），它们被分成6个数据库。我需要在这些数据库中的任何一个月获取一组样本数据。在

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

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

问题：

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

但是我得到了一个UnmappedClassError声明我的基类（所有新类都是从中派生的）没有被映射。在

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

有什么想法吗？在

如果需要，我可以在这里粘贴我的代码，但我担心它有点长。我使用实体名示例中定义的映射的map_class_to_some_table方法，并且没有修改它。在

Answer 1

我也遇到了同样的情况。我的方法如下：

class_registry = {}                                                                                                                                                                    
DbBase = declarative_base(bind=engine, class_registry=class_registry)

def get_model(modelname, tablename, metadata=DbBase.metadata):
    if modelname not in class_registry: 
        model = type(modelname, (DbBase,), dict(
            __table__ = Table(tablename, metadata, autoload=True)
        ))  
    else:
        model = class_registry[modelname]
return model

它起作用了好吧。但是@凯蒂的方法更好

Answer 2

最终放弃了所有这些，转而使用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(...