feat: initial work to add custom error messages to data contract

Author: stevenhsd

src/dve/core_engine/backends/implementations/duckdb/__init__.py

@@ -0,0 +1,15 @@
+from dve.core_engine.backends.readers import register_reader
+
+from .contract import DuckDBDataContract
+from .readers import DuckDBCSVReader, DuckDBXMLStreamReader
+from .reference_data import DuckDBRefDataLoader
+from .rules import DuckDBStepImplementations
+
+register_reader(DuckDBCSVReader)
+register_reader(DuckDBXMLStreamReader)
+
+__all__ = [
+    "DuckDBDataContract",
+    "DuckDBRefDataLoader",
+    "DuckDBStepImplementations",
+]

src/dve/core_engine/backends/implementations/duckdb/auditing.py

@@ -13,7 +13,6 @@
 )
 from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
     PYTHON_TYPE_TO_DUCKDB_TYPE,
-    PYTHON_TYPE_TO_POLARS_TYPE,
     table_exists,
 )
 from dve.core_engine.models import (
@@ -23,6 +22,7 @@
     SubmissionStatisticsRecord,
     TransferRecord,
 )
+from dve.core_engine.backends.utilities import PYTHON_TYPE_TO_POLARS_TYPE
 from dve.core_engine.type_hints import URI, ExecutorType
 
 

src/dve/core_engine/backends/implementations/duckdb/contract.py

@@ -20,13 +20,12 @@
     duckdb_read_parquet,
     duckdb_write_parquet,
     get_duckdb_type_from_annotation,
-    get_polars_type_from_annotation,
     relation_is_empty,
 )
 from dve.core_engine.backends.implementations.duckdb.types import DuckDBEntities
 from dve.core_engine.backends.metadata.contract import DataContractMetadata
 from dve.core_engine.backends.types import StageSuccessful
-from dve.core_engine.backends.utilities import stringify_model
+from dve.core_engine.backends.utilities import stringify_model, get_polars_type_from_annotation
 from dve.core_engine.message import FeedbackMessage
 from dve.core_engine.type_hints import URI, Messages
 from dve.core_engine.validation import RowValidator

src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py

@@ -15,7 +15,7 @@
 from duckdb import DuckDBPyConnection, DuckDBPyRelation
 from duckdb.typing import DuckDBPyType
 from pandas import DataFrame
-from polars.datatypes.classes import DataTypeClass as PolarsType
+
 from pydantic import BaseModel
 from typing_extensions import Annotated, get_args, get_origin, get_type_hints
 
@@ -91,20 +91,6 @@ def __call__(self):
 }
 """A mapping of Python types to the equivalent DuckDB types."""
 
-PYTHON_TYPE_TO_POLARS_TYPE: Dict[type, PolarsType] = {
-    # issue with decimal conversion at the moment...
-    str: pl.Utf8,  # type: ignore
-    int: pl.Int64,  # type: ignore
-    bool: pl.Boolean,  # type: ignore
-    float: pl.Float64,  # type: ignore
-    bytes: pl.Binary,  # type: ignore
-    date: pl.Date,  # type: ignore
-    datetime: pl.Datetime,  # type: ignore
-    Decimal: pl.Utf8,  # type: ignore
-}
-"""A mapping of Python types to the equivalent Polars types."""
-
-
 def table_exists(connection: DuckDBPyConnection, table_name: str) -> bool:
     """check if a table exists in a given DuckDBPyConnection"""
     return table_name in map(lambda x: x[0], connection.sql("SHOW TABLES").fetchall())
@@ -204,99 +190,6 @@ def get_duckdb_type_from_annotation(type_annotation: Any) -> DuckDBPyType:
             return duck_type
     raise ValueError(f"No equivalent DuckDB type for {type_annotation!r}")
 
-
-def get_polars_type_from_annotation(type_annotation: Any) -> PolarsType:
-    """Get a polars type from a Python type annotation.
-
-    Supported types  are any of the following (this definition is recursive):
-    - Supported basic Python types. These are:
-      * `str`: pl.Utf8
-      * `int`: pl.Int64
-      * `bool`: pl.Boolean
-      * `float`: pl.Float64
-      * `bytes`: pl.Binary
-      * `datetime.date`: pl.Date
-      * `datetime.datetime`: pl.Datetime
-      * `decimal.Decimal`: pl.Decimal with precision of 38 and scale of 18
-    - A list of supported types (e.g. `List[str]` or `typing.List[str]`).
-      This will return a pl.List type (variable length)
-    - A `typing.Optional` type or a `typing.Union` of the type and `None` (e.g.
-      `typing.Optional[str]`, `typing.Union[List[str], None]`). This will remove the
-      'optional' wrapper and return the inner type
-    - A subclass of `typing.TypedDict` with values typed using supported types. This
-      will parse the value types as Polars types and return a Polars Struct.
-    - A dataclass or `pydantic.main.ModelMetaClass` with values typed using supported types.
-      This will parse the field types as Polars types and return a Polars Struct.
-    - Any supported type, with a `typing_extensions.Annotated` wrapper.
-    - A `decimal.Decimal` wrapped with `typing_extensions.Annotated` with a `DecimalConfig`
-      indicating precision and scale. This will return a Polars Decimal
-      with the specfied scale and precision.
-    - A `pydantic.types.condecimal` created type.
-
-    Any `ClassVar` types within `TypedDict`s, dataclasses, or `pydantic` models will be
-    ignored.
-
-    """
-    type_origin = get_origin(type_annotation)
-
-    # An `Optional` or `Union` type, check to ensure non-heterogenity.
-    if type_origin is Union:
-        python_type = _get_non_heterogenous_type(get_args(type_annotation))
-        return get_polars_type_from_annotation(python_type)
-
-    # Type hint is e.g. `List[str]`, check to ensure non-heterogenity.
-    if type_origin is list or (isinstance(type_origin, type) and issubclass(type_origin, list)):
-        element_type = _get_non_heterogenous_type(get_args(type_annotation))
-        return pl.List(get_polars_type_from_annotation(element_type))  # type: ignore
-
-    if type_origin is Annotated:
-        python_type, *other_args = get_args(type_annotation)  # pylint: disable=unused-variable
-        return get_polars_type_from_annotation(python_type)
-    # Ensure that we have a concrete type at this point.
-    if not isinstance(type_annotation, type):
-        raise ValueError(f"Unsupported type annotation {type_annotation!r}")
-
-    if (
-        # Type hint is a dict subclass, but not dict. Possibly a `TypedDict`.
-        (issubclass(type_annotation, dict) and type_annotation is not dict)
-        # Type hint is a dataclass.
-        or is_dataclass(type_annotation)
-        # Type hint is a `pydantic` model.
-        or (type_origin is None and issubclass(type_annotation, BaseModel))
-    ):
-        fields: Dict[str, PolarsType] = {}
-        for field_name, field_annotation in get_type_hints(type_annotation).items():
-            # Technically non-string keys are disallowed, but people are bad.
-            if not isinstance(field_name, str):
-                raise ValueError(
-                    f"Dictionary/Dataclass keys must be strings, got {type_annotation!r}"
-                )  # pragma: no cover
-            if get_origin(field_annotation) is ClassVar:
-                continue
-
-            fields[field_name] = get_polars_type_from_annotation(field_annotation)
-
-        if not fields:
-            raise ValueError(
-                f"No type annotations in dict/dataclass type (got {type_annotation!r})"
-            )
-
-        return pl.Struct(fields)  # type: ignore
-
-    if type_annotation is list:
-        raise ValueError(
-            f"List must have type annotation (e.g. `List[str]`), got {type_annotation!r}"
-        )
-    if type_annotation is dict or type_origin is dict:
-        raise ValueError(f"Dict must be `typing.TypedDict` subclass, got {type_annotation!r}")
-
-    for type_ in type_annotation.mro():
-        polars_type = PYTHON_TYPE_TO_POLARS_TYPE.get(type_)
-        if polars_type:
-            return polars_type
-    raise ValueError(f"No equivalent DuckDB type for {type_annotation!r}")
-
-
 def coerce_inferred_numpy_array_to_list(pandas_df: DataFrame) -> DataFrame:
     """Function to modify numpy inferred array when cnverting from duckdb relation to
     pandas dataframe - these cause issues with pydantic models

src/dve/core_engine/backends/implementations/duckdb/readers/xml.py

@@ -10,10 +10,9 @@
 from dve.core_engine.backends.base.reader import read_function
 from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
     duckdb_write_parquet,
-    get_polars_type_from_annotation,
 )
 from dve.core_engine.backends.readers.xml import XMLStreamReader
-from dve.core_engine.backends.utilities import stringify_model
+from dve.core_engine.backends.utilities import stringify_model, get_polars_type_from_annotation
 from dve.core_engine.type_hints import URI
 
 

src/dve/core_engine/backends/implementations/duckdb/rules.py

@@ -57,7 +57,7 @@
 
 @duckdb_write_parquet
 @duckdb_read_parquet
-class DuckDBStepImplemetations(BaseStepImplementations[DuckDBPyRelation]):
+class DuckDBStepImplementations(BaseStepImplementations[DuckDBPyRelation]):
     """An implementation of transformation steps in duckdb."""
 
     def __init__(self, connection: DuckDBPyConnection, **kwargs):

src/dve/core_engine/backends/readers/csv.py

@@ -14,10 +14,7 @@
     FieldCountMismatch,
     MissingHeaderError,
 )
-from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
-    get_polars_type_from_annotation,
-)
-from dve.core_engine.backends.utilities import stringify_model
+from dve.core_engine.backends.utilities import stringify_model, get_polars_type_from_annotation
 from dve.core_engine.type_hints import EntityName
 from dve.parser.file_handling import get_content_length, open_stream
 from dve.parser.file_handling.implementations.file import file_uri_to_local_path

src/dve/core_engine/backends/readers/xml.py

@@ -11,10 +11,7 @@
 
 from dve.core_engine.backends.base.reader import BaseFileReader
 from dve.core_engine.backends.exceptions import EmptyFileError
-from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
-    get_polars_type_from_annotation,
-)
-from dve.core_engine.backends.utilities import stringify_model
+from dve.core_engine.backends.utilities import stringify_model, get_polars_type_from_annotation
 from dve.core_engine.loggers import get_logger
 from dve.core_engine.type_hints import URI, EntityName
 from dve.parser.file_handling import NonClosingTextIOWrapper, get_content_length, open_stream

src/dve/core_engine/backends/utilities.py

@@ -1,20 +1,41 @@
 """Necessary, otherwise uncategorised backend functionality."""
 
+from datetime import date, datetime
+from decimal import Decimal
 import sys
 from typing import Type
 
+from dataclasses import is_dataclass
 from pydantic import BaseModel, create_model
 
 from dve.core_engine.type_hints import Messages
+from dve.core_engine.backends.base.utilities import _get_non_heterogenous_type
+
+import polars as pl  # type: ignore
+from polars.datatypes.classes import DataTypeClass as PolarsType
+from typing import Any, ClassVar, Dict, Set, Union
 
 # We need to rely on a Python typing implementation detail in Python <= 3.7.
 if sys.version_info[:2] <= (3, 7):
     # Crimes against typing.
     from typing import _GenericAlias  # type: ignore
 
-    from typing_extensions import get_args, get_origin
+    from typing_extensions import Annotated, get_args, get_origin, get_type_hints
 else:
-    from typing import get_args, get_origin
+    from typing import Annotated, get_args, get_origin, get_type_hints
+
+PYTHON_TYPE_TO_POLARS_TYPE: Dict[type, PolarsType] = {
+    # issue with decimal conversion at the moment...
+    str: pl.Utf8,  # type: ignore
+    int: pl.Int64,  # type: ignore
+    bool: pl.Boolean,  # type: ignore
+    float: pl.Float64,  # type: ignore
+    bytes: pl.Binary,  # type: ignore
+    date: pl.Date,  # type: ignore
+    datetime: pl.Datetime,  # type: ignore
+    Decimal: pl.Utf8,  # type: ignore
+}
+"""A mapping of Python types to the equivalent Polars types."""
 
 
 def stringify_type(type_: type) -> type:
@@ -61,3 +82,94 @@ def dedup_messages(messages: Messages) -> Messages:
 
     """
     return list(dict.fromkeys(messages))
+
+def get_polars_type_from_annotation(type_annotation: Any) -> PolarsType:
+    """Get a polars type from a Python type annotation.
+
+    Supported types  are any of the following (this definition is recursive):
+    - Supported basic Python types. These are:
+      * `str`: pl.Utf8
+      * `int`: pl.Int64
+      * `bool`: pl.Boolean
+      * `float`: pl.Float64
+      * `bytes`: pl.Binary
+      * `datetime.date`: pl.Date
+      * `datetime.datetime`: pl.Datetime
+      * `decimal.Decimal`: pl.Decimal with precision of 38 and scale of 18
+    - A list of supported types (e.g. `List[str]` or `typing.List[str]`).
+      This will return a pl.List type (variable length)
+    - A `typing.Optional` type or a `typing.Union` of the type and `None` (e.g.
+      `typing.Optional[str]`, `typing.Union[List[str], None]`). This will remove the
+      'optional' wrapper and return the inner type
+    - A subclass of `typing.TypedDict` with values typed using supported types. This
+      will parse the value types as Polars types and return a Polars Struct.
+    - A dataclass or `pydantic.main.ModelMetaClass` with values typed using supported types.
+      This will parse the field types as Polars types and return a Polars Struct.
+    - Any supported type, with a `typing_extensions.Annotated` wrapper.
+    - A `decimal.Decimal` wrapped with `typing_extensions.Annotated` with a `DecimalConfig`
+      indicating precision and scale. This will return a Polars Decimal
+      with the specfied scale and precision.
+    - A `pydantic.types.condecimal` created type.
+
+    Any `ClassVar` types within `TypedDict`s, dataclasses, or `pydantic` models will be
+    ignored.
+
+    """
+    type_origin = get_origin(type_annotation)
+
+    # An `Optional` or `Union` type, check to ensure non-heterogenity.
+    if type_origin is Union:
+        python_type = _get_non_heterogenous_type(get_args(type_annotation))
+        return get_polars_type_from_annotation(python_type)
+
+    # Type hint is e.g. `List[str]`, check to ensure non-heterogenity.
+    if type_origin is list or (isinstance(type_origin, type) and issubclass(type_origin, list)):
+        element_type = _get_non_heterogenous_type(get_args(type_annotation))
+        return pl.List(get_polars_type_from_annotation(element_type))  # type: ignore
+
+    if type_origin is Annotated:
+        python_type, *other_args = get_args(type_annotation)  # pylint: disable=unused-variable
+        return get_polars_type_from_annotation(python_type)
+    # Ensure that we have a concrete type at this point.
+    if not isinstance(type_annotation, type):
+        raise ValueError(f"Unsupported type annotation {type_annotation!r}")
+
+    if (
+        # Type hint is a dict subclass, but not dict. Possibly a `TypedDict`.
+        (issubclass(type_annotation, dict) and type_annotation is not dict)
+        # Type hint is a dataclass.
+        or is_dataclass(type_annotation)
+        # Type hint is a `pydantic` model.
+        or (type_origin is None and issubclass(type_annotation, BaseModel))
+    ):
+        fields: Dict[str, PolarsType] = {}
+        for field_name, field_annotation in get_type_hints(type_annotation).items():
+            # Technically non-string keys are disallowed, but people are bad.
+            if not isinstance(field_name, str):
+                raise ValueError(
+                    f"Dictionary/Dataclass keys must be strings, got {type_annotation!r}"
+                )  # pragma: no cover
+            if get_origin(field_annotation) is ClassVar:
+                continue
+
+            fields[field_name] = get_polars_type_from_annotation(field_annotation)
+
+        if not fields:
+            raise ValueError(
+                f"No type annotations in dict/dataclass type (got {type_annotation!r})"
+            )
+
+        return pl.Struct(fields)  # type: ignore
+
+    if type_annotation is list:
+        raise ValueError(
+            f"List must have type annotation (e.g. `List[str]`), got {type_annotation!r}"
+        )
+    if type_annotation is dict or type_origin is dict:
+        raise ValueError(f"Dict must be `typing.TypedDict` subclass, got {type_annotation!r}")
+
+    for type_ in type_annotation.mro():
+        polars_type = PYTHON_TYPE_TO_POLARS_TYPE.get(type_)
+        if polars_type:
+            return polars_type
+    raise ValueError(f"No equivalent DuckDB type for {type_annotation!r}")