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 loadj(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 dumpj(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
 80

[docs]
 81class BasicSession(CouchSession):
 82    loader = BasicLoader

 83
 84

[docs]
 85class BasicPool(SessionPool):
 86    session_class = BasicSession

 87
 88

[docs]
 89class DictRegistry(DocumentRegistry):
 90    _loader = BasicLoader()
 91

[docs]
 92    def load_doc(self, cls: type[Document], blob: dict) -> Document:
 93        """
 94        Load a document
 95        """
 96        return self._loader.loadj(blob, _kind=cls)

 97

[docs]
 98    def dump_doc(self, doc: Document) -> dict:
 99        """
100        Save a document
101        """
102        return self._loader.dumpj(doc)