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