Update a (custom) dictionary in a dict-of-dict model

[doc0815]

Dear all,

I implemented a dictionary of dictionaries model. My main entity Curve has a set of dictionaries, each can be accessed by a date, i.e. it is modeled as Dict[k=date, Dict[k=float, v=float]. This date access is through a Item class. If the inner dictionary is a plain dict, I can use the dict.update() method on item to update the inner dictionary with new key/value pairs.

However, I want to use a custom dictionary as inner dict that is derived from SortedDict. Thus, I have written a getset_factory() that works for getting MyDict instances and setting them using the = assignment. But it refuses to work properly for the update() method. What am I missing in my implementation?

Best, Nicolas

from sortedcontainers import SortedDict
import datetime as dt

# sqlalchemy version: 1.3.18
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, Float, String, Date
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Session, relationship
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy

Base = declarative_base()


class MyDict(SortedDict):
    """custom sorted dictionary stuff"""

    pass


def mydict_getset_factory(*args, **kwargs):
    def getter(obj):
        print("getter called")
        return MyDict(getattr(obj, "series"))  # cast dict to MyDict

    def setter(obj, key, value):
        print("setter called")
        return setattr(obj, "series", value)

    return getter, setter


def mydict_creator(key, value):
    print("creator called")
    return Item(item_date=key, series=value)


class Curve(Base):
    __tablename__ = "curve"

    name = Column(String, primary_key=True)
    _items = relationship(
        "Item",
        collection_class=attribute_mapped_collection("item_date"),
    )
    items = association_proxy(
        target_collection="_items",
        attr="series",
        creator=mydict_creator,
        getset_factory=mydict_getset_factory,
    )


class Item(Base):
    __tablename__ = "item"

    id = Column(Integer, primary_key=True)
    curve_id = Column(Integer, ForeignKey("curve.name"), nullable=False)
    item_date = Column(Date, nullable=False)

    _series = relationship(
        "KeyValuePair",
        collection_class=attribute_mapped_collection("key"),
        cascade="all, delete, delete-orphan",
    )
    series = association_proxy(
        target_collection="_series",
        attr="value",
        creator=lambda k, v: KeyValuePair(key=k, value=v),
    )

    def __init__(self, item_date, series=None):
        self.item_date = item_date
        if series:
            self.series = series


class KeyValuePair(Base):
    __tablename__ = "key_value_pair"
    id = Column(Integer, primary_key=True)
    item_id = Column(Integer, ForeignKey("item.id"), nullable=False)
    key = Column(Float)
    value = Column(String)

    def __init__(self, key, value):
        self.key = key
        self.value = value


if __name__ == "__main__":

    engine = create_engine("sqlite://", echo=False)
    Base.metadata.create_all(engine)
    session = Session(engine)

    #
    # update on MyDict directly
    #
    a = MyDict({0.25: "A", 0.5: "B"})
    b = MyDict({0.5: "C", 1.0: "D"})
    print(a)  # MyDict({0.25: "A", 0.5: "B"})
    print(type(a))
    a.update(b)
    print(a)  # MyDict({0.25: "A", 0.5: "C", 1.0: "D"})
    print("=====")

    #
    # update on MyDict using SQLAlchemy
    #
    a = MyDict({0.25: "A", 0.5: "B"})
    b = MyDict({0.5: "C", 1.0: "D"})
    d = dt.date(2021, 12, 31)

    curve = session.query(Curve).filter_by(name="curve").one_or_none()
    if curve is None:
        curve = Curve(name="curve")

    curve.items[d] = a  # calls creator
    curve.items[d] = a  # calls setter
    tmp = curve.items[d]  # calls getter
    print(tmp)  # MyDict({0.25: 'A', 0.5: 'B'})

    print("-----")
    print(type(curve.items[d]))
    print("before call update")
    curve.items[d].update(b)  # <= DOES NOT WORK AS EXPECTED
    print("after call update")
    print(curve.items[d])  # MyDict({0.25: 'A', 0.5: 'B'})

    session.commit()

If you comment out the getset_factory in Curve.items, then the update works but I do not get my customized dictionary.

[zzzeek]

so....just looking at this and I don't deal with the assoc proxy stuff very often, the assoc proxy is a "proxy" for a real underlying collection, and the proxy itself doesnt actually implement the storage:

# without any MyDict involved
(Pdb) type(curve.items)
<class 'sqlalchemy.ext.associationproxy._AssociationDict'>
(Pdb) type(curve.items[d])
<class 'sqlalchemy.ext.associationproxy._AssociationDict'>

Since your inner dictionary is also an attribute mapped collection, that's based on plain dict so that's the part where you need to inject your sorted dictionary. that is, the dictionary you "see" from the association proxy is always going to be AssociationDict. The "creator" / "getset_factory" stuff deals with a target element that is not also another association proxy, in this case the Item class, so maybe your approach was setting the correct state (not sure) but in any case that's not the dictionary you'd get if you loaded your objects from the database, so i didnt spend much time figuring out what was going wrong there.

creating our own dict to be mapped is also a not widely used use case because it can be hard, the example is at https://docs.sqlalchemy.org/en/14/orm/collections.html#custom-dictionary-based-collections.

good news, when I just followed the docs exactly, it seems to work and I added a DB round trip plus intentionally unsorting the items to see the sort happens. This is maybe a good example we should add to examples/ .

import datetime as dt

from sortedcontainers import SortedDict

from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import Date
from sqlalchemy import Float
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.orm.collections import collection
from sqlalchemy.orm.collections import MappedCollection

# sqlalchemy version: 1.3.18

Base = declarative_base()


class MyDict(SortedDict, MappedCollection):
    """custom sorted dictionary stuff"""

    def __init__(self):
        self.keyfunc = lambda key_value_pair: key_value_pair.key
        super().__init__()

    @collection.appender
    @collection.internally_instrumented
    def set(self, value, _sa_initiator=None):
        """Add an item by value, consulting the keyfunc for the key."""

        key = self.keyfunc(value)
        self.__setitem__(key, value, _sa_initiator)

    @collection.remover
    @collection.internally_instrumented
    def remove(self, value, _sa_initiator=None):
        """Remove an item by value, consulting the keyfunc for the key."""

        key = self.keyfunc(value)
        # Let self[key] raise if key is not in this collection
        # testlib.pragma exempt:__ne__
        assert self[key] == value
        self.__delitem__(key, _sa_initiator)


def mydict_creator(key, value):
    print("creator called")
    return Item(item_date=key, series=value)


class Curve(Base):
    __tablename__ = "curve"

    name = Column(String, primary_key=True)
    _items = relationship(
        "Item",
        collection_class=attribute_mapped_collection("item_date"),
    )
    items = association_proxy(
        target_collection="_items",
        attr="series",
        creator=mydict_creator,
    )


class Item(Base):
    __tablename__ = "item"

    id = Column(Integer, primary_key=True)
    curve_id = Column(Integer, ForeignKey("curve.name"), nullable=False)
    item_date = Column(Date, nullable=False)

    _series = relationship(
        "KeyValuePair",
        collection_class=MyDict,
        cascade="all, delete, delete-orphan",
    )
    series = association_proxy(
        target_collection="_series",
        attr="value",
        creator=lambda k, v: KeyValuePair(key=k, value=v),
    )

    def __init__(self, item_date, series=None):
        self.item_date = item_date
        if series:
            self.series = series


class KeyValuePair(Base):
    __tablename__ = "key_value_pair"
    id = Column(Integer, primary_key=True)
    item_id = Column(Integer, ForeignKey("item.id"), nullable=False)
    key = Column(Float)
    value = Column(String)

    def __init__(self, key, value):
        self.key = key
        self.value = value


if __name__ == "__main__":

    engine = create_engine("sqlite://", echo=False)
    Base.metadata.create_all(engine)
    session = Session(engine)

    # I un-ordered these to see that the ordering occurs
    a = {
        0.5: "B",
        0.25: "A",
    }
    b = {1.0: "D", 0.5: "C"}
    d = dt.date(2021, 12, 31)

    curve = session.query(Curve).filter_by(name="curve").one_or_none()
    if curve is None:
        curve = Curve(name="curve")

    curve.items[d] = a
    curve.items[d] = a
    tmp = curve.items[d]
    print(tmp)
    print("-----")
    print(type(curve.items[d]))
    print("before call update")
    curve.items[d].update(b)
    print("after call update")
    print(curve.items[d])

    session.add(curve)
    session.commit()
    session.close()

    # reload to see in-python ordering again
    curve = session.query(Curve).filter_by(name="curve").one_or_none()
    print(curve.items[d])

output, the dict comes out key sorted both times

creator called
{0.25: 'A', 0.5: 'B'}
-----
<class 'sqlalchemy.ext.associationproxy._AssociationDict'>
before call update
after call update
{0.25: 'A', 0.5: 'C', 1.0: 'D'}
{0.25: 'A', 0.5: 'C', 1.0: 'D'}

[doc0815]

Dear Mike,
thank you very much for your detailed explanation and support. I also came accross using a custom MappedCollection, however I discarded my idea due to the following reason.
My dictionary has a method my_function, that makes use of the SortedDict class (I did not mention this before as I thought it is not relevant):

class MyDict(SortedDict, MappedCollection):

    def __init__(self, *args, **kwargs):
        self.keyfunc = lambda key_value_pair: key_value_pair.key
        super().__init__(*args, **kwargs)

    def my_function(self, key):
        idx = self.bisect_right(key)
        return self.peekitem(idx - 1) if idx > 0 else (key, None)

This method does a kind of forward filling, as the keys of MyDict are sorted.

I could call this method by casting the output of curve.items[d], but I tried to hide this inside the association proxy using the getset_factory above.

print(MyDict(curve.items[d]).my_function(0.4))  # 0.4 is no key

You already mentioned that curve.items[d] always is <class 'sqlalchemy.ext.associationproxy._AssociationDict'>, so there's no way to shift my cast from my application into the SQLAlchemy layer?

Best regards, Nicolas

[zzzeek]

IIUC you want this .my_function to be a method on the end _AssociaitonDict.

I think what you would do is subclass _AssociationDict, add your method, then implement the proxy_factory parameter that will return this custom collection.

Here's an example from the unit tests using _AssocaitionList:

        class CustomProxy(_AssociationList):
            def __init__(self, lazy_collection, creator, value_attr, parent):
                getter, setter = parent._default_getset(lazy_collection)
                _AssociationList.__init__(
                    self, lazy_collection, creator, getter, setter, parent
                )

        class Parent(object):
            children = association_proxy(
                "_children",
                "name",
                proxy_factory=CustomProxy,
                proxy_bulk_set=CustomProxy.extend,
            )

            def __init__(self, name):
                self.name = name

        class Child(object):
            def __init__(self, name):
                self.name = name

[doc0815]

I got it going working for my use case. Thanks a lot for your help!!!

Here's my solution and including some testing (with prints). my_function from above has been renamed into get.

from sortedcontainers import SortedDict
import datetime as dt

# sqlalchemy version: 1.3.18
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, Float, String, Date
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Session, relationship
from sqlalchemy.orm.collections import (
    attribute_mapped_collection,
    MappedCollection,
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy, _AssociationDict

Base = declarative_base()


class MyDict(SortedDict):
    """custom sorted dictionary stuff"""

    def get(self, key):
        idx = self.bisect_right(key)
        return self.peekitem(idx - 1) if idx > 0 else (key, None)


class MyDictMapCol(MyDict, MappedCollection):
    def __init__(self, *args, **kwargs):
        self.keyfunc = lambda key_value_pair: key_value_pair.key
        super().__init__(*args, **kwargs)


class CustomProxy(_AssociationDict, MyDict):
    def __init__(self, lazy_collection, creator, value_attr, parent):
        getter, setter = parent._default_getset(lazy_collection)
        setter = lambda obj, key, value: setattr(obj, value_attr, value)
        _AssociationDict.__init__(
            self, lazy_collection, creator, getter, setter, parent
        )

    def get(self, outer_key, inner_key=None):
        k, inner_dict = MyDict(self).get(outer_key)
        if inner_key:
            return MyDict(inner_dict).get(inner_key) if inner_dict else None
        else:
            return inner_dict


class Curve(Base):
    __tablename__ = "curve"

    name = Column(String, primary_key=True)
    _items = relationship(
        "Item",
        collection_class=attribute_mapped_collection("item_date"),
    )
    items = association_proxy(
        target_collection="_items",
        attr="series",
        creator=lambda key, value: Item(item_date=key, series=value),
        proxy_factory=CustomProxy,
        proxy_bulk_set=CustomProxy.update,
    )


class Item(Base):
    __tablename__ = "item"

    id = Column(Integer, primary_key=True)
    curve_id = Column(Integer, ForeignKey("curve.name"), nullable=False)
    item_date = Column(Date, nullable=False)

    _series = relationship(
        "KeyValuePair",
        # collection_class=attribute_mapped_collection("key"),
        collection_class=MyDictMapCol,
        cascade="all, delete, delete-orphan",
    )
    series = association_proxy(
        target_collection="_series",
        attr="value",
        creator=lambda k, v: KeyValuePair(key=k, value=v),
    )

    def __init__(self, item_date, series=None):
        self.item_date = item_date
        if series:
            self.series = series


class KeyValuePair(Base):
    __tablename__ = "key_value_pair"
    id = Column(Integer, primary_key=True)
    item_id = Column(Integer, ForeignKey("item.id"), nullable=False)
    key = Column(Float)
    value = Column(String)

    def __init__(self, key, value):
        self.key = key
        self.value = value


if __name__ == "__main__":

    engine = create_engine("sqlite://", echo=False)
    Base.metadata.create_all(engine)
    session = Session(engine)

    a = MyDict({0.5: "B", 0.25: "A"})
    b = MyDict({1.0: "D", 0.5: "C"})
    d = dt.date(2021, 12, 31)

    curve = session.query(Curve).filter_by(name="curve").one_or_none()
    if curve is None:
        curve = Curve(name="curve")
        session.add(curve)

    curve.items[d] = b  # create
    print(curve.items[d])  # get
    curve.items[d] = a  # set
    print(curve.items[d])

    print(type(curve.items))
    print(type(curve.items[d]))
    print("before update call")
    print(curve.items[d])
    curve.items[d].update(b)  # update
    print("after update call")
    print(curve.items[d])

    print("\nvarious access methods")
    for i in [-1, 0, 1, 5]:
        d2 = d + dt.timedelta(days=i)
        print(f"{d2}, {curve.items.get(d2)}")
        for j in range(2, 13, 2):
            print(f"({d2}, {j/10}), {curve.items.get(d2, j/10)}")
    print("")

    print(curve.items[d])
    curve.items[d].pop(1.0)  # delete
    print(curve.items[d])
    curve.items[d].setdefault(0.3, "E")  # insert
    print(curve.items[d])
    curve.items[d][0.3] = "D"  # update
    curve.items[d][0.8] = "G"  # insert
    print(curve.items[d])

    session.commit()
    session.close()

    # reload to see in-python ordering again
    curve = session.query(Curve).filter_by(name="curve").one()
    print(curve.items[d])