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
63
64
65
66
67
68
69
70
71
|
import asyncio
from typing import Any, ClassVar, Dict, Optional, Union
from pydantic import BaseModel
from beanie.exceptions import ViewWasNotInitialized
from beanie.odm.fields import Link, LinkInfo
from beanie.odm.interfaces.aggregate import AggregateInterface
from beanie.odm.interfaces.detector import DetectionInterface, ModelType
from beanie.odm.interfaces.find import FindInterface
from beanie.odm.interfaces.getters import OtherGettersInterface
from beanie.odm.settings.view import ViewSettings
class View(
BaseModel,
FindInterface,
AggregateInterface,
OtherGettersInterface,
DetectionInterface,
):
"""
What is needed:
Source collection or view
pipeline
"""
# Relations
_link_fields: ClassVar[Optional[Dict[str, LinkInfo]]] = None
# Settings
_settings: ClassVar[ViewSettings]
@classmethod
def get_settings(cls) -> ViewSettings:
"""
Get view settings, which was created on
the initialization step
:return: ViewSettings class
"""
if cls._settings is None:
raise ViewWasNotInitialized
return cls._settings
async def fetch_link(self, field: Union[str, Any]):
ref_obj = getattr(self, field, None)
if isinstance(ref_obj, Link):
value = await ref_obj.fetch(fetch_links=True)
setattr(self, field, value)
if isinstance(ref_obj, list) and ref_obj:
values = await Link.fetch_list(ref_obj, fetch_links=True)
setattr(self, field, values)
async def fetch_all_links(self):
coros = []
link_fields = self.get_link_fields()
if link_fields is not None:
for ref in link_fields.values():
coros.append(self.fetch_link(ref.field_name)) # TODO lists
await asyncio.gather(*coros)
@classmethod
def get_link_fields(cls) -> Optional[Dict[str, LinkInfo]]:
return cls._link_fields
@classmethod
def get_model_type(cls) -> ModelType:
return ModelType.View
|
