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