Source code for examples.dynamic_dict.dynamic_dict

  1. from sqlalchemy import Column
  2. from sqlalchemy import create_engine
  3. from sqlalchemy import ForeignKey
  4. from sqlalchemy import Integer
  5. from sqlalchemy import String
  6. from sqlalchemy.ext.declarative import declarative_base
  7. from sqlalchemy.orm import relationship
  8. from sqlalchemy.orm import sessionmaker
  11. class ProxyDict(object):
  12.     def __init__(self, parent, collection_name, childclass, keyname):
  13.         self.parent = parent
  14.         self.collection_name = collection_name
  15.         self.childclass = childclass
  16.         self.keyname = keyname
  18.     @property
  19.     def collection(self):
  20.         return getattr(self.parent, self.collection_name)
  22.     def keys(self):
  23.         descriptor = getattr(self.childclass, self.keyname)
  24.         return [x[0] for x in self.collection.values(descriptor)]
  26.     def __getitem__(self, key):
  27.         x = self.collection.filter_by(**{self.keyname: key}).first()
  28.         if x:
  29.             return x
  30.         else:
  31.             raise KeyError(key)
  33.     def __setitem__(self, key, value):
  34.         try:
  35.             existing = self[key]
  36.             self.collection.remove(existing)
  37.         except KeyError:
  38.             pass
  39.         self.collection.append(value)
  42. engine = create_engine("sqlite://", echo=True)
  43. Base = declarative_base(engine)
  46. class Parent(Base):
  47.     __tablename__ = "parent"
  48.     id = Column(Integer, primary_key=True)
  49.     name = Column(String(50))
  50.     _collection = relationship(
  51.         "Child", lazy="dynamic", cascade="all, delete-orphan"
  52.     )
  54.     @property
  55.     def child_map(self):
  56.         return ProxyDict(self, "_collection", Child, "key")
  59. class Child(Base):
  60.     __tablename__ = "child"
  61.     id = Column(Integer, primary_key=True)
  62.     key = Column(String(50))
  63.     parent_id = Column(Integer, ForeignKey("parent.id"))
  65.     def __repr__(self):
  66.         return "Child(key=%r)" % self.key
  69. Base.metadata.create_all()
  71. sess = sessionmaker()()
  73. p1 = Parent(name="p1")
  74. sess.add(p1)
  76. print("\n---------begin setting nodes, autoflush occurs\n")
  77. p1.child_map["k1"] = Child(key="k1")
  78. p1.child_map["k2"] = Child(key="k2")
  80. # this will autoflush the current map.
  81. # ['k1', 'k2']
  82. print("\n---------print keys - flushes first\n")
  83. print(list(p1.child_map.keys()))
  85. # k1
  86. print("\n---------print 'k1' node\n")
  87. print(p1.child_map["k1"])
  89. print("\n---------update 'k2' node - must find existing, and replace\n")
  90. p1.child_map["k2"] = Child(key="k2")
  92. print("\n---------print 'k2' key - flushes first\n")
  93. # k2
  94. print(p1.child_map["k2"])
  96. print("\n---------print all child nodes\n")
  97. # [k1, k2b]
  98. print(sess.query(Child).all())