fauna.query.models

  1from collections.abc import Mapping
  2from datetime import datetime
  3from typing import Union, Iterator, Any, Optional, List
  4
  5
  6class Page:
  7  """A class representing a Set in Fauna."""
  8
  9  def __init__(self,
 10               data: Optional[List[Any]] = None,
 11               after: Optional[str] = None):
 12    self.data = data
 13    self.after = after
 14
 15  def __repr__(self):
 16    args = []
 17    if self.data is not None:
 18      args.append(f"data={repr(self.data)}")
 19
 20    if self.after is not None:
 21      args.append(f"after={repr(self.after)}")
 22
 23    return f"{self.__class__.__name__}({','.join(args)})"
 24
 25  def __iter__(self) -> Iterator[Any]:
 26    return iter(self.data or [])
 27
 28  def __eq__(self, other):
 29    return isinstance(
 30        other, Page) and self.data == other.data and self.after == other.after
 31
 32  def __hash__(self):
 33    return hash((type(self), self.data, self.after))
 34
 35  def __ne__(self, other):
 36    return not self.__eq__(other)
 37
 38
 39class StreamToken:
 40  """A class represeting a Stream in Fauna."""
 41
 42  def __init__(self, token: str):
 43    self.token = token
 44
 45  def __eq__(self, other):
 46    return isinstance(other, StreamToken) and self.token == other.token
 47
 48  def __hash__(self):
 49    return hash(self.token)
 50
 51
 52class Module:
 53  """A class representing a Module in Fauna. Examples of modules include Collection, Math, and a user-defined
 54    collection, among others.
 55
 56    Usage:
 57
 58       dogs = Module("Dogs")
 59       query = fql("${col}.all", col=dogs)
 60    """
 61
 62  def __init__(self, name: str):
 63    self.name = name
 64
 65  def __repr__(self):
 66    return f"{self.__class__.__name__}(name={repr(self.name)})"
 67
 68  def __eq__(self, other):
 69    return isinstance(other, Module) and str(self) == str(other)
 70
 71  def __hash__(self):
 72    return hash(self.name)
 73
 74
 75class BaseReference:
 76  _collection: Module
 77
 78  @property
 79  def coll(self) -> Module:
 80    return self._collection
 81
 82  def __init__(self, coll: Union[str, Module]):
 83    if isinstance(coll, Module):
 84      self._collection = coll
 85    elif isinstance(coll, str):
 86      self._collection = Module(coll)
 87    else:
 88      raise TypeError(
 89          f"'coll' should be of type Module or str, but was {type(coll)}")
 90
 91  def __repr__(self):
 92    return f"{self.__class__.__name__}(coll={repr(self._collection)})"
 93
 94  def __eq__(self, other):
 95    return isinstance(other, type(self)) and str(self) == str(other)
 96
 97
 98class DocumentReference(BaseReference):
 99  """A class representing a reference to a :class:`Document` stored in Fauna.
100    """
101
102  @property
103  def id(self) -> str:
104    """The ID for the :class:`Document`. Valid IDs are 64-bit integers, stored as strings.
105
106        :rtype: str
107        """
108    return self._id
109
110  def __init__(self, coll: Union[str, Module], id: str):
111    super().__init__(coll)
112
113    if not isinstance(id, str):
114      raise TypeError(f"'id' should be of type str, but was {type(id)}")
115    self._id = id
116
117  def __hash__(self):
118    return hash((type(self), self._collection, self._id))
119
120  def __repr__(self):
121    return f"{self.__class__.__name__}(id={repr(self._id)},coll={repr(self._collection)})"
122
123  @staticmethod
124  def from_string(ref: str):
125    rs = ref.split(":")
126    if len(rs) != 2:
127      raise ValueError("Expects string of format <CollectionName>:<ID>")
128    return DocumentReference(rs[0], rs[1])
129
130
131class NamedDocumentReference(BaseReference):
132  """A class representing a reference to a :class:`NamedDocument` stored in Fauna.
133    """
134
135  @property
136  def name(self) -> str:
137    """The name of the :class:`NamedDocument`.
138
139        :rtype: str
140        """
141    return self._name
142
143  def __init__(self, coll: Union[str, Module], name: str):
144    super().__init__(coll)
145
146    if not isinstance(name, str):
147      raise TypeError(f"'name' should be of type str, but was {type(name)}")
148
149    self._name = name
150
151  def __hash__(self):
152    return hash((type(self), self._collection, self._name))
153
154  def __repr__(self):
155    return f"{self.__class__.__name__}(name={repr(self._name)},coll={repr(self._collection)})"
156
157
158class NullDocument:
159
160  @property
161  def cause(self) -> Optional[str]:
162    return self._cause
163
164  @property
165  def ref(self) -> Union[DocumentReference, NamedDocumentReference]:
166    return self._ref
167
168  def __init__(
169      self,
170      ref: Union[DocumentReference, NamedDocumentReference],
171      cause: Optional[str] = None,
172  ):
173    self._cause = cause
174    self._ref = ref
175
176  def __repr__(self):
177    return f"{self.__class__.__name__}(ref={repr(self.ref)},cause={repr(self._cause)})"
178
179  def __eq__(self, other):
180    if not isinstance(other, type(self)):
181      return False
182
183    return self.ref == other.ref and self.cause == other.cause
184
185  def __ne__(self, other):
186    return not self == other
187
188
189class BaseDocument(Mapping):
190  """A base document class implementing an immutable mapping.
191    """
192
193  def __init__(self, *args, **kwargs):
194    self._store = dict(*args, **kwargs)
195
196  def __getitem__(self, __k: str) -> Any:
197    return self._store[__k]
198
199  def __len__(self) -> int:
200    return len(self._store)
201
202  def __iter__(self) -> Iterator[Any]:
203    return iter(self._store)
204
205  def __eq__(self, other):
206    if not isinstance(other, type(self)):
207      return False
208
209    if len(self) != len(other):
210      return False
211
212    for k, v in self.items():
213      if k not in other:
214        return False
215      if self[k] != other[k]:
216        return False
217
218    return True
219
220  def __ne__(self, other):
221    return not self.__eq__(other)
222
223
224class Document(BaseDocument):
225  """A class representing a user document stored in Fauna.
226
227    User data should be stored directly on the map, while id, ts, and coll should only be stored on the related
228    properties. When working with a :class:`Document` in code, it should be considered immutable.
229    """
230
231  @property
232  def id(self) -> str:
233    return self._id
234
235  @property
236  def ts(self) -> datetime:
237    return self._ts
238
239  @property
240  def coll(self) -> Module:
241    return self._coll
242
243  def __init__(self,
244               id: str,
245               ts: datetime,
246               coll: Union[str, Module],
247               data: Optional[Mapping] = None):
248    if not isinstance(id, str):
249      raise TypeError(f"'id' should be of type str, but was {type(id)}")
250
251    if not isinstance(ts, datetime):
252      raise TypeError(f"'ts' should be of type datetime, but was {type(ts)}")
253
254    if not (isinstance(coll, str) or isinstance(coll, Module)):
255      raise TypeError(
256          f"'coll' should be of type Module or str, but was {type(coll)}")
257
258    if isinstance(coll, str):
259      coll = Module(coll)
260
261    self._id = id
262    self._ts = ts
263    self._coll = coll
264
265    super().__init__(data or {})
266
267  def __eq__(self, other):
268    return type(self) == type(other) \
269        and self.id == other.id \
270        and self.coll == other.coll \
271        and self.ts == other.ts \
272        and super().__eq__(other)
273
274  def __ne__(self, other):
275    return not self.__eq__(other)
276
277  def __repr__(self):
278    kvs = ",".join([f"{repr(k)}:{repr(v)}" for k, v in self.items()])
279
280    return f"{self.__class__.__name__}(" \
281           f"id={repr(self.id)}," \
282           f"coll={repr(self.coll)}," \
283           f"ts={repr(self.ts)}," \
284           f"data={{{kvs}}})"
285
286
287class NamedDocument(BaseDocument):
288  """A class representing a named document stored in Fauna. Examples of named documents include Collection
289    definitions, Index definitions, and Roles, among others.
290
291    When working with a :class:`NamedDocument` in code, it should be considered immutable.
292    """
293
294  @property
295  def name(self) -> str:
296    return self._name
297
298  @property
299  def ts(self) -> datetime:
300    return self._ts
301
302  @property
303  def coll(self) -> Module:
304    return self._coll
305
306  def __init__(self,
307               name: str,
308               ts: datetime,
309               coll: Union[Module, str],
310               data: Optional[Mapping] = None):
311    if not isinstance(name, str):
312      raise TypeError(f"'name' should be of type str, but was {type(name)}")
313
314    if not isinstance(ts, datetime):
315      raise TypeError(f"'ts' should be of type datetime, but was {type(ts)}")
316
317    if not (isinstance(coll, str) or isinstance(coll, Module)):
318      raise TypeError(
319          f"'coll' should be of type Module or str, but was {type(coll)}")
320
321    if isinstance(coll, str):
322      coll = Module(coll)
323
324    self._name = name
325    self._ts = ts
326    self._coll = coll
327
328    super().__init__(data or {})
329
330  def __eq__(self, other):
331    return type(self) == type(other) \
332        and self.name == other.name \
333        and self.coll == other.coll \
334        and self.ts == other.ts \
335        and super().__eq__(other)
336
337  def __ne__(self, other):
338    return not self.__eq__(other)
339
340  def __repr__(self):
341    kvs = ",".join([f"{repr(k)}:{repr(v)}" for k, v in self.items()])
342
343    return f"{self.__class__.__name__}(" \
344           f"name={repr(self.name)}," \
345           f"coll={repr(self.coll)}," \
346           f"ts={repr(self.ts)}," \
347           f"data={{{kvs}}})"