Coverage for / home / runner / work / viur-core / viur-core / viur / src / viur / core / db / types.py: 86%
73 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-13 14:41 +0000
1"""
2The constants, global variables and container classes used in the datastore api
3"""
4from __future__ import annotations
6import datetime
7import enum
8import itertools
9import typing as t
10from contextvars import ContextVar
11from dataclasses import dataclass, field
12from ..config import conf
14from google.cloud.datastore import Entity as Datastore_entity, Key as Datastore_key
16KEY_SPECIAL_PROPERTY = "__key__"
17"""The property name pointing to an entities key in a query"""
19DATASTORE_BASE_TYPES = t.Union[None, str, int, float, bool, datetime.datetime, datetime.date, datetime.time, "Key"]
20"""Types that can be used in a datastore query"""
22current_db_access_log: ContextVar[t.Optional[set[t.Union[Key, str]]]] = ContextVar("Database-Accesslog", default=None)
23"""If set to a set for the current thread/request, we'll log all entities / kinds accessed"""
25"""The current projectID, which can't be imported from transport.py"""
28class SortOrder(enum.Enum):
29 Ascending = 1
30 """Sort A->Z"""
31 Descending = 2
32 """Sort Z->A"""
33 InvertedAscending = 3
34 """Fetch Z->A, then flip the results (useful in pagination to go from a start cursor backwards)"""
35 InvertedDescending = 4
36 """Fetch A->Z, then flip the results (useful in pagination)"""
39class Key(Datastore_key):
40 """
41 The python representation of one datastore key. Unlike the original implementation, we don't store a
42 reference to the project the key lives in. This is always expected to be the current project as ViUR
43 does not support accessing data in multiple projects.
44 """
46 def __init__(self, *path_args, project: str | None = None, **kwargs):
47 # Convert digit-only id_or_name attributes to int
48 # See https://github.com/viur-framework/viur-core/issues/1636
49 new_path_args = []
50 for pair in itertools.batched(path_args, 2):
51 try:
52 kind, id_or_name = pair
53 except ValueError: # it's a incomplete key
54 new_path_args.append(pair[0])
55 continue
56 if isinstance(id_or_name, str) and id_or_name.isdigit():
57 id_or_name = int(id_or_name)
58 new_path_args.extend((kind, id_or_name))
60 if project is None: 60 ↛ 64line 60 didn't jump to line 64 because the condition on line 60 was always true
61 from .transport import __client__ # noqa: E402 # import works only here because circular imports
62 project = __client__.project
64 super().__init__(*new_path_args, project=project, **kwargs)
66 def __str__(self):
67 return self.to_legacy_urlsafe().decode("ASCII")
69 '''
70 def __repr__(self):
71 return "<viur.datastore.Key %s/%s, parent=%s>" % (self.kind, self.id_or_name, self.parent)
73 def __hash__(self):
74 return hash("%s.%s.%s" % (self.kind, self.id, self.name))
76 def __eq__(self, other):
77 return isinstance(other, Key) and self.kind == other.kind and self.id == other.id and self.name == other.name \
78 and self.parent == other.parent
80 @staticmethod
81 def _parse_path(path_args):
82 """Parses positional arguments into key path with kinds and IDs.
84 :type path_args: tuple
85 :param path_args: A tuple from positional arguments. Should be
86 alternating list of kinds (string) and ID/name
87 parts (int or string).
89 :rtype: :class:`list` of :class:`dict`
90 :returns: A list of key parts with kind and ID or name set.
91 :raises: :class:`ValueError` if there are no ``path_args``, if one of
92 the kinds is not a string or if one of the IDs/names is not
93 a string or an integer.
94 """
95 if len(path_args) == 0:
96 raise ValueError("Key path must not be empty.")
98 kind_list = path_args[::2]
99 id_or_name_list = path_args[1::2]
100 # Dummy sentinel value to pad incomplete key to even length path.
101 partial_ending = object()
102 if len(path_args) % 2 == 1:
103 id_or_name_list += (partial_ending,)
105 result = []
106 for kind, id_or_name in zip(kind_list, id_or_name_list):
107 curr_key_part = {}
108 if isinstance(kind, str):
109 curr_key_part["kind"] = kind
110 else:
111 raise ValueError(kind, "Kind was not a string.")
113 if isinstance(id_or_name, str):
114 if (id_or_name.isdigit()): # !!! VIUR
115 curr_key_part["id"] = int(id_or_name)
116 else:
117 curr_key_part["name"] = id_or_name
119 elif isinstance(id_or_name, int):
120 curr_key_part["id"] = id_or_name
121 elif id_or_name is not partial_ending:
122 raise ValueError(id_or_name, "ID/name was not a string or integer.")
124 result.append(curr_key_part)
125 return result
127 @classmethod
128 def from_legacy_urlsafe(cls, strKey: str) -> Key:
129 """
130 Parses the string representation generated by :meth:to_legacy_urlsafe into a new Key object
131 :param strKey: The string key to parse
132 :return: The new Key object constructed from the string key
133 """
134 urlsafe = strKey.encode("ASCII")
135 padding = b"=" * (-len(urlsafe) % 4)
136 urlsafe += padding
137 raw_bytes = base64.urlsafe_b64decode(urlsafe)
138 reference = _app_engine_key_pb2.Reference()
139 reference.ParseFromString(raw_bytes)
140 resultKey = None
141 for elem in reference.path.element:
142 resultKey = Key(elem.type, elem.id or elem.name, parent=resultKey)
143 return resultKey
144 '''
147class Entity(Datastore_entity):
148 """
149 The python representation of one datastore entity. The values of this entity are stored inside this dictionary,
150 while the meta-data (it's key, the list of properties excluded from indexing and our version) as property values.
151 """
153 def __init__(
154 self,
155 key: t.Optional[Key] = None,
156 exclude_from_indexes: t.Optional[list[str]] = None,
157 ) -> None:
158 super().__init__(key, exclude_from_indexes or [])
159 if not (key is None or isinstance(key, Key)):
160 raise ValueError(f"key must be a Key-Object (or None for an embedded entity). Got {key!r} ({type(key)})")
163KeyType: t.TypeAlias = Key | str | int
164"""
165Alias that describes a key-type.
166"""
168TOrders: t.TypeAlias = list[tuple[str, SortOrder]]
169TFilters: t.TypeAlias = dict[str, DATASTORE_BASE_TYPES | list[DATASTORE_BASE_TYPES]]
172@dataclass
173class QueryDefinition:
174 """
175 A single Query that will be run against the datastore.
176 """
178 kind: t.Optional[str]
179 """The datastore kind to run the query on. Can be None for kindles queries."""
181 filters: TFilters
182 """A dictionary of constrains to apply to the query."""
184 orders: t.Optional[TOrders]
185 """The list of fields to sort the results by."""
187 distinct: t.Optional[list[str]] = None
188 """If set, a list of fields that we should return distinct values of"""
190 limit: int = field(init=False)
191 """The maximum amount of entities that should be returned"""
193 startCursor: t.Optional[str] = None
194 """If set, we'll only return entities that appear after this cursor in the index."""
196 endCursor: t.Optional[str] = None
197 """If set, we'll only return entities up to this cursor in the index."""
199 currentCursor: t.Optional[str] = None
200 """Will be set after this query has been run, pointing after the last entity returned"""
202 def __post_init__(self):
203 self.limit = conf.db.query_default_limit