How do I create a multi-level dictionary backed by a single table in SQLAlchemy?

Question

I have a table that looks like this:

values = Table('values',
...
Column('person_id', Unicode(34), ForeignKey('persons.guid')),
Column('type', Text, nullable=False),
Column('name', Text, nullable=False),
Column('params', HSTORE, default={}),
UniqueConstraint('person_id', 'type', 'name'),

I'd like to configure a relationship on the Person object such that I can do

p = Person()
p.values['a type']['a name'] = {'my': 'params'}

and have it work as expected.

I've looked at http://docs.sqlalchemy.org/en/latest/_modules/examples/association/dict_of_sets_with_default.html, which looks similar to what I want, including the nice defaultdict-like functionality. It looks like that goes through an association table, though, and offers a many-to-many relation. I just want a one-to-many relation, and want each Person to have many possible name values under each of many type values.

Answer 1

to do this all in one table means you're just writing on top of an existing collection. There are several methods of trying to achieve this but all of them require writing a very elaborate nested collection class. One way is to write this nested collection class, such that it persists all changes to a simple list-based relationship. That's an easy way to go but it's not terribly efficient as you have to either rewrite the backing list on change or otherwise search through it. Another way is to try to tango with the collection decorators. Part of this is below. Note we are writing this totally custom so we just skip the association proxy...every setitem/getitem has to be accounted for. Below I'm using just a string for the value, that can be replaced with the hstore.

from sqlalchemy.orm.collections import MappedCollection, \
                    collection_adapter, collection

class MySubMagicDict(dict):
    def __init__(self, parent, key):
        self.parent = parent
        self.key = key

    def __setitem__(self, key, value):
        self.parent.__setitem__(self.key,
                            Value(type=self.key, name=key, value=value))

class MyMagicDict(MappedCollection):
    def __missing__(self, key):
        d = MySubMagicDict(self, key)
        dict.__setitem__(self, key, d)
        return d

    def __getitem__(self, key):
        d = MySubMagicDict(self, key)
        d.update(
            (v.name, v.value) for v in dict.__getitem__(self, key).values()
        )
        return d

    @collection.internally_instrumented
    def __setitem__(self, key, value, _sa_initiator=None):

        if _sa_initiator is not False:
            executor = collection_adapter(self)
            if executor:
                value = executor.fire_append_event(value, _sa_initiator)

        dict.__setitem__(
            dict.__getitem__(self, key), value.name, value)


    @collection.converter
    def _convert(self, dictlike):
        for incoming_key, value in dictlike.items():
            for name, subvalue in value.items():
                yield Value(type=incoming_key, name=name, value=subvalue)

    @collection.iterator
    def _all_items(self):
        for value in self.values():
            for subvalue in value.values():
                yield subvalue

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

values = Table('values', Base.metadata,
    Column("id", Integer, primary_key=True),
    Column('person_id', Integer, ForeignKey('persons.id')),
    Column('type', Text, nullable=False),
    Column('name', Text, nullable=False),
    Column('value', String)
)
class Value(Base):
    __table__ = values

class Person(Base):
    __tablename__ = 'persons'

    id = Column(Integer, primary_key=True)
    values = relationship("Value", collection_class=
                            lambda: MyMagicDict(lambda item: item.type),
                            cascade="all, delete-orphan"
                        )

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

s = Session(e)

p = Person(values={
    't1': {
        'n1': 't1n1',
        'n2': 't1n2'
    },
    't2': {
        'n1': 't2n1',
        'n2': 't2n2'
    }
})
s.add(p)
s.commit()

assert p.values['t2']['n2'] == 't2n2'
assert p.values['t1']['n1'] == 't1n1'

p.values['t1']['n2'] = 't1n2mod'

s.commit()

assert p.values['t1']['n2'] == 't1n2mod'
assert p.values['t1']['n1'] == 't1n1'