Added ProxyDict

CHANGES.rst

@@ -4,6 +4,12 @@ Changelog
 Here you can see the full list of changes between each SQLAlchemy-Utils release.
 
 
+0.12.0 (2013-05-08)
+^^^^^^^^^^^^^^^^^^^
+
+- Added ProxyDict
+
+
 0.11.0 (2013-05-08)
 ^^^^^^^^^^^^^^^^^^^
 

requirements-dev.txt

@@ -3,3 +3,4 @@ pytest==2.2.3
 Pygments==1.2
 Jinja2==2.3
 docutils>=0.10
+flexmock>=0.9.7

setup.py

@@ -24,7 +24,7 @@ class PyTest(Command):
 
 setup(
     name='SQLAlchemy-Utils',
-    version='0.11.0',
+    version='0.12.0',
     url='https://github.com/kvesteri/sqlalchemy-utils',
     license='BSD',
     author='Konsta Vesterinen',

sqlalchemy_utils/__init__.py

@@ -1,6 +1,7 @@
 from .functions import sort_query, defer_except, escape_like
 from .listeners import coercion_listener
 from .merge import merge, Merger
+from .proxy_dict import ProxyDict
 from .types import (
     ColorType,
     EmailType,
@@ -34,6 +35,7 @@ __all__ = (
     NumberRangeType,
     PhoneNumber,
     PhoneNumberType,
+    ProxyDict,
     ScalarListType,
     ScalarListException,
 )

sqlalchemy_utils/proxy_dict.py

@@ -0,0 +1,53 @@
+import sqlalchemy as sa
+
+
+class ProxyDict(object):
+    def __init__(self, parent, collection_name, child_class, key_name):
+        self.parent = parent
+        self.collection_name = collection_name
+        self.child_class = child_class
+        self.key_name = key_name
+        self.cache = {}
+
+    @property
+    def collection(self):
+        return getattr(self.parent, self.collection_name)
+
+    def keys(self):
+        descriptor = getattr(self.child_class, self.key_name)
+        return [x[0] for x in self.collection.values(descriptor)]
+
+    def __contains__(self, key):
+        try:
+            return key in self.cache or self[key]
+        except KeyError:
+            return False
+
+    def fetch(self, key):
+        return self.collection.filter_by(**{self.key_name: key}).first()
+
+    def __getitem__(self, key):
+        if key in self.cache:
+            return self.cache[key]
+
+        session = sa.orm.object_session(self.parent)
+        if not session or not sa.orm.util.has_identity(self.parent):
+            value = self.child_class(**{self.key_name: key})
+            self.collection.append(value)
+        else:
+            value = self.fetch(key)
+            if not value:
+                value = self.child_class(**{self.key_name: key})
+                self.collection.append(value)
+
+        self.cache[key] = value
+        return value
+
+    def __setitem__(self, key, value):
+        try:
+            existing = self[key]
+            self.collection.remove(existing)
+        except KeyError:
+            pass
+        self.collection.append(value)
+        self.cache[key] = value

tests/test_proxy_dict.py

@@ -0,0 +1,85 @@
+from flexmock import flexmock
+import sqlalchemy as sa
+from sqlalchemy_utils import ProxyDict
+from tests import TestCase
+
+
+class TestProxyDict(TestCase):
+    def create_models(self):
+        class Article(self.Base):
+            __tablename__ = 'article'
+
+            id = sa.Column(sa.Integer, autoincrement=True, primary_key=True)
+            description = sa.Column(sa.UnicodeText)
+            _translations = sa.orm.relationship(
+                'ArticleTranslation',
+                lazy='dynamic',
+                cascade='all, delete-orphan',
+                passive_deletes=True,
+                backref=sa.orm.backref('parent'),
+            )
+
+            @property
+            def translations(self):
+                try:
+                    return self.proxied_translations
+                except AttributeError:
+                    self.proxied_translations = ProxyDict(
+                        self,
+                        '_translations',
+                        ArticleTranslation,
+                        'locale'
+                    )
+                return self.proxied_translations
+
+        class ArticleTranslation(self.Base):
+            __tablename__ = 'article_translation'
+
+            id = sa.Column(
+                sa.Integer,
+                sa.ForeignKey(Article.id),
+                autoincrement=True,
+                primary_key=True
+            )
+            locale = sa.Column(sa.String(10), primary_key=True)
+            name = sa.Column(sa.UnicodeText)
+
+        self.Article = Article
+        self.ArticleTranslation = ArticleTranslation
+
+    def test_access_key_for_pending_parent(self):
+        article = self.Article()
+        self.session.add(article)
+        assert article.translations['en']
+
+    def test_access_key_for_transient_parent(self):
+        article = self.Article()
+        assert article.translations['en']
+
+    def test_cache(self):
+        article = self.Article()
+        (
+            flexmock(ProxyDict)
+            .should_receive('fetch')
+            .once()
+        )
+        self.session.add(article)
+        self.session.commit()
+        article.translations['en']
+        article.translations['en']
+
+    def test_set_updates_cache(self):
+        article = self.Article()
+        (
+            flexmock(ProxyDict)
+            .should_receive('fetch')
+            .once()
+        )
+        self.session.add(article)
+        self.session.commit()
+        article.translations['en']
+        article.translations['en'] = self.ArticleTranslation(
+            locale='en',
+            name=u'something'
+        )
+        article.translations['en']