fauna.query.models

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