python - SQLAlchemy Classical Mapping Model to sharded Postgres databases -
the situation:
i have set of 12 tables (representing data month) sharded across 6 databases. need sample set of data across of these databases given month.
why used classical mappping model rather declarative model:
i require access 1 of 12 types of table gathering sample of data single given month each time code run. classical mapping model allows me dynamically define table name want map @ run time, rather create mappings 12 tables across 6 databases believe required declarative.
the problem:
i trying follow entity_name example given here mapping month data class each of tables given month on 6 different databases.
but getting unmappedclasserror
stating base class, new classes derived from, 'is not mapped'.
so on trying initialise 1 of new mapped tables type: <class '__main__.db1month1'>
reporting unmappedclasserror: class 'audit.db.orm.mappedclasses.monthdata' not mapped
.
any ideas?
if needed can paste in code here i'm worried it's little long. using map_class_to_some_table
method defined in entity_name example mappings , haven't altered it.
ended scrapping , following this shardedsession example instead.
my final class looks this:
class shardsessionmanager(object): def __init__(self, month): self.month = month #step1: database engines self.engines = {} 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 databases within 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(...
Comments
Post a Comment