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
|
"""
This file contains mechanisms for logging development data to files, SQL databases, and other formats.
"""
import json
from datetime import datetime
from typing import Any, Dict
import aiohttp
from .create_async_task import create_async_task
from .logging import logger
from .paths import getDevDataFilePath
class DevDataLogger:
user_token: str = None
data_server_url: str = None
def setup(self, user_token: str = None, data_server_url: str = None):
self.user_token = user_token
self.data_server_url = data_server_url
def _to_data_server(self, table_name: str, data: Dict[str, Any]):
async def _async_helper(self, table_name: str, data: Dict[str, Any]):
if self.user_token is None or self.data_server_url is None:
return
async with aiohttp.ClientSession() as session:
await session.post(
f"{self.data_server_url}/event",
headers={"Authorization": f"Bearer {self.user_token}"},
json={
"table_name": table_name,
"data": data,
"user_token": self.user_token,
},
)
create_async_task(
_async_helper(self, table_name, data),
lambda e: logger.warning(f"Failed to send dev data: {e}"),
)
def _static_columns(self):
return {
"user_token": self.user_token or "NO_USER_TOKEN",
"timestamp": datetime.now().isoformat(),
}
def _to_local(self, table_name: str, data: Dict[str, Any]):
filepath = getDevDataFilePath(table_name)
with open(filepath, "a") as f:
json_line = json.dumps(data)
f.write(f"{json_line}\n")
def capture(self, table_name: str, data: Dict[str, Any]):
try:
data = {**self._static_columns(), **data}
self._to_data_server(table_name, data)
self._to_local(table_name, data)
except Exception as e:
logger.warning(f"Failed to capture dev data: {e}")
dev_data_logger = DevDataLogger()
|
