Source code for chaise.dictful

  1"""
  2A basic version of chaise that just uses enriched dictionaries as the documents.
  3
  4:class:`BasicLoader`, :class:`BasicSession`, and :class:`BasicPool` do not do
  5any migration handling.
  6
  7:class:`DictRegistry` should be paired with :class:`~chaise.CouchSession` and
  8:class:`~chaise.SessionPool` if you want some basic document type handling.
  9"""
 10
 11from . import DocumentRegistry, CouchSession, SessionPool
 12
 13

[docs]
 14class Document(dict):
 15    """
 16    A dictionary with some CouchDB attributes
 17    """
 18
 19    #: Document ID
 20    id: str | None = None
 21
 22    #: Document revision
 23    rev: str | None = None
 24
 25    #: Has the document been deleted? (ie, is this a tombstone?)
 26    deleted: bool = False
 27
 28    #: Attachment information, if requested
 29    attachments: dict | None = None
 30
 31    #: List of conflicts, if requested
 32    conflicts: list | None = None
 33
 34    # List of deleted conflicts, if requested
 35    deleted_conflicts: list | None = None
 36
 37    #:
 38    local_seq: str | None = None
 39
 40    #:
 41    revs_info: list | None = None
 42
 43    #:
 44    revisions: dict | None = None
 45
 46    def __repr__(self):
 47        return f"<{type(self).__name__} {self.id!r}@{self.rev!r} {' '.join(f'{k}={v!r}' for k,v in self.items())}>"

 48
 49

[docs]
 50class BasicLoader:
 51    """
 52    Provides loading without worrying about types or migrations.
 53    """
 54

[docs]
 55    def load_from_blob(self, blob: dict, _kind=Document) -> Document:
 56        doc = _kind()
 57        doc.id = blob.pop("_id", None)
 58        doc.rev = blob.pop("_rev", None)
 59        doc.deleted = blob.pop("_deleted", False)
 60        doc.attachments = blob.pop("_attachments", None)
 61        doc.conflicts = blob.pop("_conflicts", None)
 62        doc.deleted_conflicts = blob.pop("_deleted_conflicts", None)
 63        doc.local_seq = blob.pop("_local_seq", None)
 64        doc.revs_info = blob.pop("_revs_info", None)
 65        doc.revisions = blob.pop("_revisions", None)
 66        doc.update(blob)
 67        return doc

 68

[docs]
 69    def dump_to_blob(self, doc: Document) -> dict:
 70        return (
 71            doc
 72            | ({"_id": doc.id} if doc.id is not None else {})
 73            | ({"_rev": doc.rev} if doc.rev is not None else {})
 74            | {
 75                "_deleted": doc.deleted,
 76                # The rest of it is informational not editable directly
 77            }
 78        )

 79

[docs]
 80    def update_doc(self, doc, **fields):
 81        assert all(k.startswith("_") for k in fields)
 82        for k, v in fields.items():
 83            setattr(doc, k.removeprefix("_"), v)


 84
 85

[docs]
 86class BasicSession(CouchSession):
 87    loader = BasicLoader

 88
 89

[docs]
 90class BasicPool(SessionPool):
 91    session_class = BasicSession

 92
 93

[docs]
 94class DictRegistry(DocumentRegistry):
 95    _loader = BasicLoader()
 96

[docs]
 97    def load_doc(self, cls: type[Document], blob: dict) -> Document:
 98        """
 99        Load a document
100        """
101        return self._loader.load_from_blob(blob, _kind=cls)

102

[docs]
103    def dump_doc(self, doc: Document) -> dict:
104        """
105        Save a document
106        """
107        return self._loader.dump_to_blob(doc)

108

[docs]
109    def update_doc(self, doc, **fields):
110        self._loader.update_doc(doc, **fields)