FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
databricks-sql-python/src/databricks/sql/telemetry/utils.py at main · databricks/databricks-sql-python · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
databricks
/
databricks-sql-python
Public
Notifications
You must be signed in to change notification settings
Fork
147
Star
233
Code
Issues
100
Pull requests
78
Actions
Projects
Wiki
Security and quality
0
Insights
Additional navigation options
Code
Issues
Pull requests
Actions
Projects
Wiki
Security and quality
Insights
Expand file tree
Breadcrumbs
databricks-sql-python
/
src
/
databricks
/
sql
/
telemetry
/
utils.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
69 lines (55 loc) · 1.97 KB
Breadcrumbs
databricks-sql-python
/
src
/
databricks
/
sql
/
telemetry
/
utils.py
Copy path
File metadata and controls
69 lines (55 loc) · 1.97 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
import
json
from
enum
import
Enum
from
dataclasses
import
asdict
,
is_dataclass
from
abc
import
ABC
,
abstractmethod
import
logging
logger
=
logging
.
getLogger
(
__name__
)
class
BaseTelemetryClient
(
ABC
):
"""
Base class for telemetry clients.
It is used to define the interface for telemetry clients.
"""
@
abstractmethod
def
export_initial_telemetry_log
(
self
,
driver_connection_params
,
user_agent
):
logger
.
debug
(
"subclass must implement export_initial_telemetry_log"
)
pass
@
abstractmethod
def
export_failure_log
(
self
,
error_name
,
error_message
):
logger
.
debug
(
"subclass must implement export_failure_log"
)
pass
@
abstractmethod
def
export_latency_log
(
self
,
latency_ms
,
sql_execution_event
,
sql_statement_id
):
logger
.
debug
(
"subclass must implement export_latency_log"
)
pass
@
abstractmethod
def
close
(
self
):
logger
.
debug
(
"subclass must implement close"
)
pass
class
JsonSerializableMixin
:
"""Mixin class to provide JSON serialization capabilities to dataclasses."""
def
to_json
(
self
)
->
str
:
"""
Convert the object to a JSON string, excluding None values.
Handles Enum serialization and filters out None values from the output.
"""
if
not
is_dataclass
(
self
):
raise
TypeError
(
f"
{
self
.
__class__
.
__name__
}
must be a dataclass to use JsonSerializableMixin"
)
return
json
.
dumps
(
asdict
(
self
,
dict_factory
=
lambda
data
: {
k
:
v
for
k
,
v
in
data
if
v
is
not
None
},
),
cls
=
EnumEncoder
,
)
class
EnumEncoder
(
json
.
JSONEncoder
):
"""
Custom JSON encoder to handle Enum values.
This is used to convert Enum values to their string representations.
Default JSON encoder raises a TypeError for Enums.
"""
def
default
(
self
,
obj
):
if
isinstance
(
obj
,
Enum
):
return
obj
.
value
return
super
().
default
(
obj
)
Back
|
FazBrowse Home
|
New Git URL