My problem was: I had a number of databases generated in different machines and I wanted to query them as if they were one, using the database the data came from as a field while querying and while showing results. The databases are SQLite3 files, generated using SQLAlchemy in a Python program.
I solved this by using SQLAlchemy, which was good because I could use the same ORM mapping that the program used. I noticed that the Horizontal Sharding SQLAlchemy extension would fit well the problem, although not perfectly. I had to make some changes in some classes of this extension, and now it works fine.
It was possible to filter the data using database as a criteria, but I couldn't get the database information from each line of a query result. I made a simple patch to SQLAlchemy, which wasn't likely to be introduced in the distribution, but worked for me, and sent it to its bug tracker. The change was included in SQLAlchemy in a very different fashion, as expected, but since I'm using the released version from SQLAlchemy, I kept on using my version of the patch. I don't want to do direct changes in SQLAlchemy source code, so I made the change in my program:
class ShardedSessionShardId(ShardedSession):
def __init__(self, *args, **kwargs):
super(ShardedSessionShardId, self).__init__(*args, **kwargs)
self._query_cls = ShardedQueryShardId
class ShardedQueryShardId(ShardedQuery):
def _execute_and_instances(self, context):
if self._shard_id is not None:
result = self.session.connection(
mapper=self._mapper_zero(),
shard_id=self._shard_id).execute(context.statement, self._params)
news = list(self.instances(result, context))
for new in news:
new.shard_id = self._shard_id
return iter(news)
else:
partial = []
for shard_id in self.query_chooser(self):
result = self.session.connection(
mapper=self._mapper_zero(),
shard_id=shard_id).execute(context.statement, self._params)
news = list(self.instances(result, context))
for new in news:
new.shard_id = shard_id
partial = partial + news
# if some kind of in memory 'sorting'
# were done, this is where it would happen
return iter(partial)
create_session = sessionmaker(class_=ShardedSessionShardId)
Another problem is that I had to make each result be included in the query, even if two results from different DBs have the same primary key. I achieved this by changing two classes: WeakInstanceDict, and Mapper. For using the new WeakInstanceDict, I had again to change the ShardedSession variation:
class WeakInstanceDictNoIdentity(WeakInstanceDict):
def add(self, state):
# if state.key in self:
# if dict.__getitem__(self, state.key) is not state:
# raise AssertionError("A conflicting state is already "
# "present in the identity map for key %r"
# % (state.key, ))
# else:
dict.__setitem__(self, state.key, state)
self._manage_incoming_state(state)
class ShardedSessionShardId(ShardedSession):
def __init__(self, *args, **kwargs):
super(ShardedSessionShardId, self).__init__(*args, **kwargs)
self._query_cls = ShardedQueryShardId
self._identity_cls = WeakInstanceDictNoIdentity
self.identity_map = self._identity_cls()
To start using the new Mapper, I simply replaced each call to mapper with MapperNoIdentity:
class MapperNoIdentity(Mapper):
def _instance_processor(self, context, path, adapter,
polymorphic_from=None, extension=None,
only_load_props=None, refresh_state=None,
polymorphic_discriminator=None):
"""Produce a mapper level row processor callable
which processes rows into mapped instances."""
pk_cols = self.primary_key
if polymorphic_from or refresh_state:
polymorphic_on = None
else:
if polymorphic_discriminator is not None:
polymorphic_on = polymorphic_discriminator
else:
polymorphic_on = self.polymorphic_on
polymorphic_instances = util.PopulateDict(
self._configure_subclass_mapper(
context, path, adapter)
)
version_id_col = self.version_id_col
if adapter:
pk_cols = [adapter.columns[c] for c in pk_cols]
if polymorphic_on is not None:
polymorphic_on = adapter.columns[polymorphic_on]
if version_id_col is not None:
version_id_col = adapter.columns[version_id_col]
identity_class = self._identity_class
def identity_key(row):
return identity_class, tuple([row[column] for column in pk_cols])
new_populators = []
existing_populators = []
load_path = context.query._current_path + path
def populate_state(state, dict_, row, isnew, only_load_props):
if isnew:
if context.propagate_options:
state.load_options = context.propagate_options
if state.load_options:
state.load_path = load_path
if not new_populators:
new_populators[:], existing_populators[:] = \
self._populators(context, path, row,
adapter)
if isnew:
populators = new_populators
else:
populators = existing_populators
if only_load_props:
populators = [p for p in populators
if p[0] in only_load_props]
for key, populator in populators:
populator(state, dict_, row)
session_identity_map = context.session.identity_map
if not extension:
extension = self.extension
translate_row = extension.get('translate_row', None)
create_instance = extension.get('create_instance', None)
populate_instance = extension.get('populate_instance', None)
append_result = extension.get('append_result', None)
populate_existing = context.populate_existing or self.always_refresh
if self.allow_partial_pks:
is_not_primary_key = _none_set.issuperset
else:
is_not_primary_key = _none_set.issubset
def _instance(row, result):
if translate_row:
ret = translate_row(self, context, row)
if ret is not EXT_CONTINUE:
row = ret
if polymorphic_on is not None:
discriminator = row[polymorphic_on]
if discriminator is not None:
_instance = polymorphic_instances[discriminator]
if _instance:
return _instance(row, result)
# determine identity key
if refresh_state:
identitykey = refresh_state.key
if identitykey is None:
# super-rare condition; a refresh is being called
# on a non-instance-key instance; this is meant to only
# occur within a flush()
identitykey = self._identity_key_from_state(refresh_state)
else:
identitykey = identity_key(row)
# instance = session_identity_map.get(identitykey)
# if instance is not None:
# state = attributes.instance_state(instance)
# dict_ = attributes.instance_dict(instance)
# isnew = state.runid != context.runid
# currentload = not isnew
# loaded_instance = False
# if not currentload and \
# version_id_col is not None and \
# context.version_check and \
# self._get_state_attr_by_column(
# state,
# dict_,
# self.version_id_col) != \
# row[version_id_col]:
# raise orm_exc.ConcurrentModificationError(
# "Instance '%s' version of %s does not match %s"
# % (state_str(state),
# self._get_state_attr_by_column(
# state, dict_,
# self.version_id_col),
# row[version_id_col]))
# elif refresh_state:
if refresh_state:
# out of band refresh_state detected (i.e. its not in the
# session.identity_map) honor it anyway. this can happen
# if a _get() occurs within save_obj(), such as
# when eager_defaults is True.
state = refresh_state
instance = state.obj()
dict_ = attributes.instance_dict(instance)
isnew = state.runid != context.runid
currentload = True
loaded_instance = False
else:
# check for non-NULL values in the primary key columns,
# else no entity is returned for the row
if is_not_primary_key(identitykey[1]):
return None
isnew = True
currentload = True
loaded_instance = True
if create_instance:
instance = create_instance(self,
context,
row, self.class_)
if instance is EXT_CONTINUE:
instance = self.class_manager.new_instance()
else:
manager = attributes.manager_of_class(
instance.__class__)
# TODO: if manager is None, raise a friendly error
# about returning instances of unmapped types
manager.setup_instance(instance)
else:
instance = self.class_manager.new_instance()
dict_ = attributes.instance_dict(instance)
state = attributes.instance_state(instance)
state.key = identitykey
# manually adding instance to session. for a complete add,
# session._finalize_loaded() must be called.
state.session_id = context.session.hash_key
session_identity_map.add(state)
if currentload or populate_existing:
if isnew:
state.runid = context.runid
context.progress[state] = dict_
if not populate_instance or \
populate_instance(self, context, row, instance,
only_load_props=only_load_props,
instancekey=identitykey, isnew=isnew) is \
EXT_CONTINUE:
populate_state(state, dict_, row, isnew, only_load_props)
else:
# populate attributes on non-loading instances which have
# been expired
# TODO: apply eager loads to un-lazy loaded collections ?
if state in context.partials or state.unloaded:
if state in context.partials:
isnew = False
(d_, attrs) = context.partials[state]
else:
isnew = True
attrs = state.unloaded
# allow query.instances to commit the subset of attrs
context.partials[state] = (dict_, attrs)
if not populate_instance or \
populate_instance(self, context, row, instance,
only_load_props=attrs,
instancekey=identitykey, isnew=isnew) is \
EXT_CONTINUE:
populate_state(state, dict_, row, isnew, attrs)
if loaded_instance:
state._run_on_load(instance)
if result is not None and \
(not append_result or
append_result(self, context, row, instance,
result, instancekey=identitykey,
isnew=isnew)
is EXT_CONTINUE):
result.append(instance)
return instance
return _instance
I had to include some auxiliary definitions to make the rewrites work:
_none_set = frozenset([None])
_runid = 1L
_id_lock = util.threading.Lock()
def _new_runid():
global _runid
_id_lock.acquire()
try:
_runid += 1
return _runid
finally:
_id_lock.release()
It would be good to be able to set these identity requirements as a parameter.
My last problem was selecting more than one database to search. setshard only worked for one, so I created a new field in query, called shards, and checked for it on querychooser:
def query_chooser(query):
try:
return query.shards
except AttributeError:
pass
return tcs.keys()
So, when I want to look only in a list of shards, I set this field. I'm aware that this is not a recommended python idiom, but, well, it works fine.