Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
from dve.core_engine.backends.utilities import DEFAULT_ISO_FORMATS, datetime_format_to_regex
from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME
from dve.core_engine.type_hints import URI, EntityName
from dve.metadata_parser.utilities import resilient_get
from dve.parser.file_handling.service import LocalFilesystemImplementation, _get_implementation


Expand Down Expand Up @@ -451,23 +452,27 @@ def get_duckdb_cast_statement_from_annotation(
raise ValueError(f"dict must be `typing.TypedDict` subclass, got {type_annotation!r}")

for type_ in type_annotation.mro():
_date_format: str = getattr( # type: ignore
type_, "DATE_FORMAT", DEFAULT_ISO_FORMATS.get(type_, DEFAULT_ISO_FORMATS.get(datetime))
)
dt_cast_statement = rf"CASE WHEN REGEXP_FULL_MATCH(TRIM({quoted_name}), '{datetime_format_to_regex(_date_format)}') THEN TRY_STRPTIME(TRIM({quoted_name}), '{_date_format}') ELSE NULL END" # pylint: disable=C0301

# datetime is subclass of date, so needs to be handled first
if issubclass(type_, datetime):
stmt = rf"TRY_CAST({dt_cast_statement} as TIMESTAMP)"
return stmt
if issubclass(type_, date):
stmt = rf"TRY_CAST({dt_cast_statement} as DATE)"
return stmt
if issubclass(type_, time):
stmt = rf"TRY_CAST({dt_cast_statement} as TIME)"
return stmt
duck_type = get_duckdb_type_from_annotation(type_)
if duck_type:
stmt = f"TRIM({quoted_name})"
return _cast_as_ddb_type(stmt, type_) if parent_element else stmt
if issubclass(type_, (date, time)):
_date_format: str = resilient_get(
type_, "DATE_FORMAT", "TIME_FORMAT"
) or DEFAULT_ISO_FORMATS.get(
type_, DEFAULT_ISO_FORMATS.get(datetime)
) # type: ignore
dt_cast_statement = rf"CASE WHEN REGEXP_FULL_MATCH(TRIM({quoted_name}), '{datetime_format_to_regex(_date_format)}') THEN TRY_STRPTIME(TRIM({quoted_name}), '{_date_format}') ELSE NULL END" # pylint: disable=C0301

# datetime is subclass of date, so needs to be handled first
if issubclass(type_, datetime):
stmt = rf"TRY_CAST({dt_cast_statement} as TIMESTAMP)"
return stmt
if issubclass(type_, date):
stmt = rf"TRY_CAST({dt_cast_statement} as DATE)"
return stmt
if issubclass(type_, time):
stmt = rf"TRY_CAST({dt_cast_statement} as TIME)"
return stmt
else:
duck_type = get_duckdb_type_from_annotation(type_)
if duck_type:
stmt = f"TRIM({quoted_name})"
return _cast_as_ddb_type(stmt, type_) if parent_element else stmt
raise ValueError(f"No equivalent DuckDB type for {type_annotation!r}")
48 changes: 25 additions & 23 deletions src/dve/core_engine/backends/implementations/spark/spark_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -593,29 +593,31 @@ def get_spark_cast_statement_from_annotation(
raise ValueError(f"dict must be `typing.TypedDict` subclass, got {type_annotation!r}")

for type_ in type_annotation.mro():
_date_format: str = getattr( # type: ignore
type_,
"DATE_FORMAT",
DEFAULT_ISO_FORMATS.get(type_, DEFAULT_ISO_FORMATS.get(dt.datetime)),
)

# pylint: disable=C0301
dt_cast_statement = f"CASE WHEN REGEXP(TRIM({quoted_name}), '{datetime_format_to_regex(_date_format)}') THEN TRY_TO_TIMESTAMP(TRIM({quoted_name}), \"{python_to_java_datetime_format(_date_format)}\") ELSE NULL END" # pylint: disable=C0301
# datetime is subclass of date, so needs to be handled first
if issubclass(type_, dt.datetime):
return (
_cast_as_spark_type(dt_cast_statement, type_)
if parent_element
else dt_cast_statement
)
if issubclass(type_, dt.date):
return (
_cast_as_spark_type(dt_cast_statement, type_)
if parent_element
else dt_cast_statement
_date_format: str = getattr( # type: ignore
type_,
"DATE_FORMAT",

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are you not checking for "TIME_FORMAT" here as well?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

version of spark being used doesnt support time datatype at the moment - better to not try to cast to a type that doesn't exist

DEFAULT_ISO_FORMATS.get(type_, DEFAULT_ISO_FORMATS.get(dt.datetime)),
)
spark_type = get_type_from_annotation(type_)
if spark_type:
stmt = f"TRIM({quoted_name})"
return _cast_as_spark_type(stmt, type_) if parent_element else stmt

# pylint: disable=C0301
dt_cast_statement = f"CASE WHEN REGEXP(TRIM({quoted_name}), '{datetime_format_to_regex(_date_format)}') THEN TRY_TO_TIMESTAMP(TRIM({quoted_name}), \"{python_to_java_datetime_format(_date_format)}\") ELSE NULL END" # pylint: disable=C0301
# datetime is subclass of date, so needs to be handled first
if issubclass(type_, dt.datetime):
return (
_cast_as_spark_type(dt_cast_statement, type_)
if parent_element
else dt_cast_statement
)
if issubclass(type_, dt.date):
return (
_cast_as_spark_type(dt_cast_statement, type_)
if parent_element
else dt_cast_statement
)
else:
spark_type = get_type_from_annotation(type_)
if spark_type:
stmt = f"TRIM({quoted_name})"
return _cast_as_spark_type(stmt, type_) if parent_element else stmt
raise ValueError(f"No equivalent Spark type for {type_annotation!r}")
20 changes: 20 additions & 0 deletions src/dve/metadata_parser/utilities.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,3 +52,23 @@ def chain_get(
return result

raise exc.TypeNotFoundError(f"Callable or type ({item!r}) not found")


def resilient_get(item: object, *attribute_names: str) -> Any:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
def resilient_get(item: object, *attribute_names: str) -> Any:
def resilient_get(item: object, *attribute_names: tuple[str]) -> Any:

?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

changed docs instead - using *args to gather into a tuple

"""Given a number of attribute names, try to get attribute value
sequentially. Returns the first value found, and if no attributes found
returns None.

Args:
item (object): The object to obtain attributes from (where possible)
attribute_names (str): The attribute names to search for

Returns:
Any: The first found attribute, otherwise None
"""
for attr in attribute_names:
try:
return getattr(item, attr)
except AttributeError:
continue
return None
24 changes: 24 additions & 0 deletions tests/test_parser/test_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
import pytest
from dve.metadata_parser.utilities import resilient_get

class MyParent:
cls_attr = "hello"
def __init__(self, my_attr:str, another_attr:str):
self.my_attr = my_attr
self.another_attr = another_attr

class MyObject(MyParent):
sub_attr = "bye"
def __init__(self, extra_attr:int):
self.extra_attr = extra_attr
super().__init__("from", "child")


@pytest.mark.parametrize("obj,attrs,expected", [(MyParent, ("cls_attr",), "hello"),
(MyObject, ("cls_attr", "sub_attr"), "hello"),
(MyObject, ("sub_attr", "cls_attr"), "bye"),
(MyParent, ("my_attr",), None),
(MyParent("this", "test"), ("extra_attr", "my_attr"), "this"),
(MyObject("this"), ("daft_attr", "another_daft_attr", "yet_another", "another_attr"), "child")])
def test_resilient_get(obj, attrs, expected):
assert resilient_get(obj, *attrs) == expected
Loading