forked from amundsen-io/amundsendatabuilder
-
Notifications
You must be signed in to change notification settings - Fork 0
/
dashboard_usage.py
96 lines (83 loc) · 3.56 KB
/
dashboard_usage.py
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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
# Copyright Contributors to the Amundsen project.
# SPDX-License-Identifier: Apache-2.0
import logging
from typing import Optional, Dict, Any, Union, Iterator
from databuilder.models.dashboard.dashboard_metadata import DashboardMetadata
from databuilder.models.neo4j_csv_serde import (
Neo4jCsvSerializable, RELATION_START_KEY, RELATION_END_KEY, RELATION_START_LABEL,
RELATION_END_LABEL, RELATION_TYPE, RELATION_REVERSE_TYPE)
from databuilder.models.usage.usage_constants import (
READ_RELATION_TYPE, READ_REVERSE_RELATION_TYPE, READ_RELATION_COUNT_PROPERTY
)
from databuilder.models.user import User
LOGGER = logging.getLogger(__name__)
class DashboardUsage(Neo4jCsvSerializable):
"""
A model that encapsulate Dashboard usage between Dashboard and User
"""
def __init__(self,
dashboard_group_id: Optional[str],
dashboard_id: Optional[str],
email: str,
view_count: int,
should_create_user_node: Optional[bool] = False,
product: Optional[str] = '',
cluster: Optional[str] = 'gold',
**kwargs: Any
) -> None:
"""
:param dashboard_group_id:
:param dashboard_id:
:param email:
:param view_count:
:param should_create_user_node: Enable this if it is fine to create/update User node with only with email
address. Please be advised that other fields will be emptied. Current use case is to create anonymous user.
For example, Mode dashboard does not provide which user viewed the dashboard and anonymous user can be used
to show the usage.
:param product:
:param cluster:
:param kwargs:
"""
self._dashboard_group_id = dashboard_group_id
self._dashboard_id = dashboard_id
self._email = email
self._view_count = view_count
self._product = product
self._cluster = cluster
self._user_model = User(email=email)
self._should_create_user_node = bool(should_create_user_node)
self._relation_iterator = self._create_relation_iterator()
def create_next_node(self) -> Union[Dict[str, Any], None]:
if self._should_create_user_node:
return self._user_model.create_next_node()
return None
def create_next_relation(self) -> Union[Dict[str, Any], None]:
try:
return next(self._relation_iterator)
except StopIteration:
return None
def _create_relation_iterator(self) -> Iterator[Dict[str, Any]]:
yield {
RELATION_START_LABEL: DashboardMetadata.DASHBOARD_NODE_LABEL,
RELATION_END_LABEL: User.USER_NODE_LABEL,
RELATION_START_KEY: DashboardMetadata.DASHBOARD_KEY_FORMAT.format(
product=self._product,
cluster=self._cluster,
dashboard_group=self._dashboard_group_id,
dashboard_name=self._dashboard_id
),
RELATION_END_KEY: User.get_user_model_key(email=self._email),
RELATION_TYPE: READ_REVERSE_RELATION_TYPE,
RELATION_REVERSE_TYPE: READ_RELATION_TYPE,
READ_RELATION_COUNT_PROPERTY: self._view_count
}
def __repr__(self) -> str:
return 'DashboardUsage({!r}, {!r}, {!r}, {!r}, {!r}, {!r}, {!r})'.format(
self._dashboard_group_id,
self._dashboard_id,
self._email,
self._view_count,
self._should_create_user_node,
self._product,
self._cluster
)