SQLAlchemy Classical Mapping Model to sharded Post

Question

The situation:

I have a set of 12 tables (representing data by month) that are sharded across 6 databases. I need to get a sample set of data across any of these databases for any given month.

Why I used Classical Mappping Model rather than Declarative Model:

I only require access to 1 of the 12 types of table as I will only be gathering a sample of data for a single given month each time this code is run. The Classical Mapping Model allows me to dynamically define the table name I want to map to at run time, rather than create mappings for all 12 tables across the 6 databases as I believe would be required with Declarative.

The problem:

I am trying to follow the entity_name example given here mapping my month data class to each of the tables for the given month on the 6 different databases.

But am getting an UnmappedClassError stating that my base class, which all new classes are derived from, 'is not mapped'.

So on trying to initialise one of my new mapped tables type: <class '__main__.db1month1'> it is reporting UnmappedClassError: Class 'audit.db.orm.mappedclasses.MonthData' is not mapped.

Any ideas?

If needed I can paste in my code here but I'm worried it's a little long. I am using the map_class_to_some_table method defined in the entity_name example for the mappings and haven't altered it.

Answer 1

I meet same situation. My method below:

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

It work well.But @Katie ‘s method is better

Answer 2

Ended up scrapping all that and following this ShardedSession example instead.

My final class looks something like this:

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(...