1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
from mongoengine.errors import NotRegistered
__all__ = ("UPDATE_OPERATORS", "get_document", "_document_registry")
UPDATE_OPERATORS = {
"set",
"unset",
"inc",
"dec",
"mul",
"pop",
"push",
"push_all",
"pull",
"pull_all",
"add_to_set",
"set_on_insert",
"min",
"max",
"rename",
}
_document_registry = {}
def get_document(name):
"""Get a registered Document class by name."""
doc = _document_registry.get(name, None)
if not doc:
# Possible old style name
single_end = name.split(".")[-1]
compound_end = ".%s" % single_end
possible_match = [
k for k in _document_registry if k.endswith(compound_end) or k == single_end
]
if len(possible_match) == 1:
doc = _document_registry.get(possible_match.pop(), None)
if not doc:
raise NotRegistered(
"""
`%s` has not been registered in the document registry.
Importing the document class automatically registers it, has it
been imported?
""".strip()
% name
)
return doc
def _get_documents_by_db(connection_alias, default_connection_alias):
"""Get all registered Documents class attached to a given database"""
def get_doc_alias(doc_cls):
return doc_cls._meta.get("db_alias", default_connection_alias)
return [
doc_cls
for doc_cls in _document_registry.values()
if get_doc_alias(doc_cls) == connection_alias
]
|
