Классическая модель сопоставления SQLAlchemy для сегментированных баз данных Postgres

Ситуация:

У меня есть набор из 12 таблиц (представляющих данные по месяцам), которые разбиты на 6 баз данных. Мне нужно получить набор данных по любой из этих баз данных за любой месяц.

Почему я использовал классическую модель сопоставления, а не декларативную модель:

Мне требуется только доступ к 1 из 12 типов таблиц, так как я буду собирать только выборку данных за один данный месяц каждый раз, когда запускается этот код. Классическая модель сопоставления позволяет мне динамически определять имя таблицы, с которой я хочу сопоставить во время выполнения, вместо того, чтобы создавать сопоставления для всех 12 таблиц в 6 базах данных, что, как я полагаю, потребуется для декларативного.

Эта проблема:

Я пытаюсь следовать приведенному здесь примеру entity_name, сопоставляя мой класс данных месяца с каждой таблицей за данный месяц в 6 разных базах данных.

Но получаю UnmappedClassError заявив, что мой базовый класс, из которого получены все новые классы, "не сопоставлен".

Итак, при попытке инициализировать одну из моих новых сопоставленных таблиц type: <class '__main__.db1month1'> это отчетность UnmappedClassError: Class 'audit.db.orm.mappedclasses.MonthData' is not mapped,

Есть идеи?

При необходимости я могу вставить свой код здесь, но я боюсь, что это немного долго. Я использую map_class_to_some_table метод, определенный в примере entity_name для отображений и не изменивший его.

Источник

2 ответа

Решение

Закончил все это списанием и следовал этому примеру ShardedSession.

Мой последний класс выглядит примерно так:

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(...
Источник

Я встречаю ту же ситуацию. Мой метод ниже:

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

Это работает хорошо. Но метод @Katie лучше

Источник
Другие вопросы по тегам