chore: Remove more redundant code in utils/core (#25986)
This commit is contained in:
parent
aee94b39ba
commit
d20b60edd4
|
|
@ -72,7 +72,7 @@ from sqlalchemy.dialects.mysql import MEDIUMTEXT
|
|||
from sqlalchemy.engine import Connection, Engine
|
||||
from sqlalchemy.engine.reflection import Inspector
|
||||
from sqlalchemy.sql.type_api import Variant
|
||||
from sqlalchemy.types import TEXT, TypeDecorator, TypeEngine
|
||||
from sqlalchemy.types import TypeEngine
|
||||
from typing_extensions import TypeGuard
|
||||
|
||||
from superset.constants import (
|
||||
|
|
@ -122,18 +122,6 @@ InputType = TypeVar("InputType") # pylint: disable=invalid-name
|
|||
ADHOC_FILTERS_REGEX = re.compile("^adhoc_filters")
|
||||
|
||||
|
||||
class LenientEnum(Enum):
|
||||
"""Enums with a `get` method that convert a enum value to `Enum` if it is a
|
||||
valid value."""
|
||||
|
||||
@classmethod
|
||||
def get(cls, value: Any) -> Any:
|
||||
try:
|
||||
return super().__new__(cls, value)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
class AdhocMetricExpressionType(StrEnum):
|
||||
SIMPLE = "SIMPLE"
|
||||
SQL = "SQL"
|
||||
|
|
@ -280,15 +268,6 @@ class PostProcessingContributionOrientation(StrEnum):
|
|||
COLUMN = "column"
|
||||
|
||||
|
||||
class QueryMode(str, LenientEnum):
|
||||
"""
|
||||
Whether the query runs on aggregate or returns raw records
|
||||
"""
|
||||
|
||||
RAW = "raw"
|
||||
AGGREGATE = "aggregate"
|
||||
|
||||
|
||||
class QuerySource(Enum):
|
||||
"""
|
||||
The source of a SQL query.
|
||||
|
|
@ -454,22 +433,6 @@ class DashboardEncoder(json.JSONEncoder):
|
|||
return json.JSONEncoder(sort_keys=True).default(o)
|
||||
|
||||
|
||||
class JSONEncodedDict(TypeDecorator): # pylint: disable=abstract-method
|
||||
"""Represents an immutable structure as a json-encoded string."""
|
||||
|
||||
impl = TEXT
|
||||
|
||||
def process_bind_param(
|
||||
self, value: dict[Any, Any] | None, dialect: str
|
||||
) -> str | None:
|
||||
return json.dumps(value) if value is not None else None
|
||||
|
||||
def process_result_value(
|
||||
self, value: str | None, dialect: str
|
||||
) -> dict[Any, Any] | None:
|
||||
return json.loads(value) if value is not None else None
|
||||
|
||||
|
||||
def format_timedelta(time_delta: timedelta) -> str:
|
||||
"""
|
||||
Ensures negative time deltas are easily interpreted by humans
|
||||
|
|
|
|||
|
|
@ -59,7 +59,6 @@ from superset.utils.core import (
|
|||
get_stacktrace,
|
||||
json_int_dttm_ser,
|
||||
json_iso_dttm_ser,
|
||||
JSONEncodedDict,
|
||||
merge_extra_filters,
|
||||
merge_extra_form_data,
|
||||
merge_request_params,
|
||||
|
|
@ -583,15 +582,6 @@ class TestUtils(SupersetTestCase):
|
|||
"-16 days, 4:03:00",
|
||||
)
|
||||
|
||||
def test_json_encoded_obj(self):
|
||||
obj = {"a": 5, "b": ["a", "g", 5]}
|
||||
val = '{"a": 5, "b": ["a", "g", 5]}'
|
||||
jsonObj = JSONEncodedDict()
|
||||
resp = jsonObj.process_bind_param(obj, "dialect")
|
||||
self.assertIn('"a": 5', resp)
|
||||
self.assertIn('"b": ["a", "g", 5]', resp)
|
||||
self.assertEqual(jsonObj.process_result_value(val, "dialect"), obj)
|
||||
|
||||
def test_validate_json(self):
|
||||
valid = '{"a": 5, "b": [1, 5, ["g", "h"]]}'
|
||||
self.assertIsNone(validate_json(valid))
|
||||
|
|
|
|||
Loading…
Reference in New Issue