Add example for fluttergram integration

examples/fluttergram/__init__.py

@@ -0,0 +1 @@
+from . import domain_models

examples/fluttergram/domain_models/__init__.py

@@ -0,0 +1,2 @@
+from .user import User
+from .post import Post

examples/fluttergram/domain_models/post.py

@@ -0,0 +1,17 @@
+from flask_boiler import schema, fields, domain_model, factory
+
+
+class PostSchema(schema.Schema):
+    owner_id = fields.Str()
+
+
+class PostBase(domain_model.DomainModel):
+
+    class Meta:
+        collection_name = "insta_posts"
+
+
+class Post(PostBase):
+
+    class Meta:
+        schema_cls = PostSchema

examples/fluttergram/domain_models/user.py

@@ -0,0 +1,17 @@
+from flask_boiler import schema, fields, domain_model, factory
+
+
+class UserSchema(schema.Schema):
+    followers = fields.Raw()
+
+
+class UserBase(domain_model.DomainModel):
+
+    class Meta:
+        collection_name = "users"
+
+
+class User(UserBase):
+
+    class Meta:
+        schema_cls = UserSchema

examples/fluttergram/tests/fixtures.py

@@ -0,0 +1,37 @@
+import pytest
+
+from examples.fluttergram import domain_models
+
+
+@pytest.fixture
+def users(request, CTX):
+    tijuana = domain_models.User.new(doc_id="tijuana")
+    tijuana.followers = {"thomasina": True, }
+    tijuana.save()
+
+    thomasina = domain_models.User.new(doc_id="thomasina")
+    thomasina.followers = dict()
+    thomasina.save()
+
+    def fin():
+        tijuana.delete()
+        thomasina.delete()
+
+    request.addfinalizer(fin)
+
+    return [tijuana, thomasina]
+
+
+@pytest.fixture
+def posts(users, request, CTX):
+    p = domain_models.Post.new(doc_id="tj_t", owner_id="tijuana", _remainder={
+        "hello": "world"
+    })
+    p.save()
+
+    def fin():
+        p.delete()
+
+    request.addfinalizer(fin)
+
+    return [p]

examples/fluttergram/tests/test_delta_dav.py

@@ -0,0 +1,100 @@
+import time
+
+import pytest
+from google.cloud.firestore_v1 import Watch, DocumentSnapshot, \
+    DocumentReference, Query
+
+from examples.fluttergram.domain_models import Post, User
+
+from flask_boiler.view import DocumentAsView
+from flask_boiler.view_mediator_dav import ViewMediatorDAV, \
+    ViewMediatorDeltaDAV
+from flask_boiler.view_model import ViewModel
+from flask_boiler import view_mediator, utils, schema, fields
+# Import the fixtures used by fixtures
+from tests.fixtures import CTX, setup_app
+from .fixtures import users, posts
+from flask_boiler.context import Context
+
+
+class PostDAVSchema(schema.Schema):
+    post = fields.Embedded(dump_only=True)
+
+
+class PostDAV(DocumentAsView):
+    class Meta:
+        schema_cls = PostDAVSchema
+
+    @property
+    def post(self):
+        return self.business_properties["post"]
+
+    @classmethod
+    def new(cls, *args, consumer_id, post_id, **kwargs):
+        ref = User._get_collection().document(consumer_id).collection("feed") \
+            .document(post_id)
+
+        struct = dict()
+        struct["consumer"] = (User, consumer_id)
+        struct["post"] = (Post, post_id)
+
+        obj = cls.get(*args, struct_d=struct, doc_ref=ref, **kwargs)
+
+        return obj
+
+
+class FeedEntryViewMediatorDeltaDAV(ViewMediatorDeltaDAV):
+
+    def __init__(self, *args, post_cls=None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.post_cls = post_cls
+
+    def _get_query_and_on_snapshot(self):
+        query = Query(parent=self.post_cls._get_collection())
+
+        def on_snapshot(snapshots, changes, timestamp):
+            for change, snapshot in zip(changes, snapshots):
+                if change.type.name == 'ADDED':
+                    post = utils.snapshot_to_obj(
+                        snapshot,
+                        super_cls=self.post_cls
+                    )
+
+                    producer_id = post.owner_id
+                    producer = User.get(doc_id=producer_id)
+                    followers = list()
+                    for key, val in producer.followers.items():
+                        if val is True:
+                            followers.append(key)
+
+                    for consumer_id in followers:
+                        obj = self.view_model_cls.new(
+                            consumer_id=consumer_id,
+                            post_id=post.doc_id,
+                            once=False,
+                            f_notify=self.notify
+                        )
+
+        return query, on_snapshot
+
+
+def test_start(users, posts):
+    mediator = FeedEntryViewMediatorDeltaDAV(
+        view_model_cls=PostDAV,
+        post_cls=Post,
+    )
+
+    mediator.start()
+
+    time.sleep(10)
+
+    ref = User._get_collection().document("thomasina").collection("feed") \
+        .document(posts[0].doc_id)
+
+    assert ref.get().to_dict() == {'obj_type': 'PostDAV',
+                                   'post': {'ownerId': 'tijuana',
+                                            'doc_ref': 'insta_posts/tj_t',
+                                            'obj_type': 'Post',
+                                            'hello': 'world',
+                                            'doc_id': 'tj_t'},
+                                   'doc_ref': 'users/thomasina/feed/tj_t'}