1"""
2Integration for attrs/cattrs
3"""
4
5import typing
6from typing import AbstractSet
7import warnings
8
9import attrs
10import cattrs.converters
11import cattrs.gen
12
13try:
14 # ujson is preferred, since muffin&c can also use it
15 from cattrs.preconf.ujson import configure_converter
16except ImportError:
17 from cattrs.preconf.json import configure_converter
18# Omitting orjson, even though it's a preconf, because I'm not confident it's a
19# drop-in equivalent to (u)json
20
21from . import DocumentRegistry
22
23
24# All implementations exhibit the conversions:
25# * bytes are wrapped in base85
26# * dates & datetimes are ISO 8601
27#: The converter used when talking to CouchDB.
28converter = cattrs.converters.Converter(
29 unstruct_collection_overrides={
30 AbstractSet: list,
31 }
32)
33configure_converter(converter)
34
35
36class classprop:
37 """
38 Like @property, but for class attributes
39
40 :meta private:
41 """
42
43 def __init__(self, factory: typing.Callable[[type], typing.Any]):
44 self._factory = factory
45 self.__doc__ = factory.__doc__
46
47 def __get__(self, instance, owner):
48 return self._factory(owner)
49
50 def __set__(self, instance, value):
51 raise AttributeError("Cannot set a classprop")
52
53 def __set_name__(self, owner, name):
54 self.__objclass__ = owner
55
56
57class DocMeta(type):
58 """
59 Defines an attrs class as a subclass instead of a decorator
60
61 :meta private:
62 """
63
64 def __new__(cls, name, bases, dict, **kwds):
65 sub = super().__new__(cls, name, bases, dict)
66 dbid = kwds.pop("dbid", None)
67 assert not kwds.get("slots", None)
68 if "__attrs_attrs__" not in dict: # prevents recursion
69 sub = attrs.define(**kwds)(sub)
70
71 # Register the class with chaise
72 if sub.__module__ != globals()["__name__"] and dbid is None:
73 warnings.warn(f"No dbid given for {sub!r}")
74 if dbid is not None and sub._Document__parent is not None:
75 sub._Document__parent.document(dbid)(sub)
76
77 # Register the class with cattrs
78 converter.register_unstructure_hook(
79 sub,
80 cattrs.gen.make_dict_unstructure_fn(
81 sub,
82 converter,
83 _id=cattrs.gen.override(omit=True),
84 _rev=cattrs.gen.override(omit=True),
85 _deleted=cattrs.gen.override(omit_if_default=True),
86 _attachments=cattrs.gen.override(omit=True),
87 _conflicts=cattrs.gen.override(omit=True),
88 _deleted_conflicts=cattrs.gen.override(omit=True),
89 _local_seq=cattrs.gen.override(omit=True),
90 _revs_info=cattrs.gen.override(omit=True),
91 _revisions=cattrs.gen.override(omit=True),
92 ),
93 )
94 return sub
95
96
[docs]
97class Document(metaclass=DocMeta, slots=False, frozen=False):
98 """
99 The parent class for all documents that get saved to Couch.
100
101 Do not inherit from directly; use :class:`AttrsRegistry.Document` instead.
102 """
103
104 __parent: typing.ClassVar[type | None] = None
105
106 #: Document ID
107 #:
108 #: :meta public:
109 _id: str | None = attrs.field(default=None, kw_only=True, alias="_id")
110
111 #: Document revision
112 #:
113 #: :meta public:
114 _rev: str | None = attrs.field(default=None, kw_only=True, alias="_rev")
115
116 #: Has the document been deleted? (ie, is this a tombstone?)
117 #:
118 #: :meta public:
119 _deleted: bool = attrs.field(default=False, kw_only=True, alias="_deleted")
120
121 #: Attachment information, if requested
122 #:
123 #: :meta public:
124 _attachments: dict | None = attrs.field(
125 default=None, kw_only=True, alias="_attachments"
126 )
127
128 #: List of alternate document versions, if requested
129 #:
130 #: :meta public:
131 _conflicts: list | None = attrs.field(
132 default=None, kw_only=True, alias="_conflicts"
133 )
134
135 #: List of deleted conflicts, if requested
136 #:
137 #: :meta public:
138 _deleted_conflicts: list | None = attrs.field(
139 default=None, kw_only=True, alias="_deleted_conflicts"
140 )
141
142 #:
143 #: :meta public:
144 _local_seq: str | None = attrs.field(default=None, kw_only=True, alias="_local_seq")
145
146 #:
147 #: :meta public:
148 _revs_info: list | None = attrs.field(
149 default=None, kw_only=True, alias="_revs_info"
150 )
151
152 #:
153 #: :meta public:
154 _revisions: dict | None = attrs.field(
155 default=None, kw_only=True, alias="_revisions"
156 )
157
158
[docs]
159class AttrsRegistry(DocumentRegistry):
[docs]
160 @classprop
161 def Document(cls) -> type[Document]:
162 """
163 Document superclass.
164
165 Handles:
166
167 * Making an attrs
168 * Registering with chaise
169 * Serialization concerns
170
171 Me sure to include the ``dbid`` keyword argument::
172
173 class MyDoc(AttrsRegistry.Document, dbid="mydoc"): ...
174 """
175
176 # This is some shenanigans because names
177 class Document(globals()["Document"]):
178 __doc__ = vars(AttrsRegistry)["Document"].__doc__
179 __parent = cls
180
181 return Document
182
183 def load_doc(self, cls: type, blob: dict):
184 """
185 :meta private:
186 """
187 return converter.structure(blob, cls)
188
189 def dump_doc(self, doc) -> dict:
190 """
191 :meta private:
192 """
193 return converter.unstructure(doc)