From befc2a391bbf32e3f24db0956adf92d1c7830acc Mon Sep 17 00:00:00 2001 From: stevenhsd <56357022+stevenhsd@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:22:41 +0100 Subject: [PATCH 1/2] feat: strip null records in readers at file transformation stage --- src/dve/core_engine/backends/base/reader.py | 30 ++- .../implementations/duckdb/duckdb_helpers.py | 13 +- .../implementations/duckdb/readers/csv.py | 6 +- .../implementations/duckdb/readers/json.py | 3 +- .../implementations/duckdb/readers/xml.py | 3 +- .../implementations/duckdb/utilities.py | 2 +- .../implementations/spark/readers/csv.py | 3 +- .../implementations/spark/readers/json.py | 2 + .../implementations/spark/readers/xml.py | 5 +- .../implementations/spark/spark_helpers.py | 7 + src/dve/core_engine/backends/utilities.py | 7 + .../test_backends/fixtures.py | 24 ++ .../test_backends/test_readers/fixtures.py | 230 ++++++++++++++++++ .../test_backends/test_readers/test_csv.py | 28 ++- .../test_readers/test_duckdb/test_ddb_csv.py | 158 ++++++------ .../test_readers/test_duckdb/test_ddb_json.py | 46 +--- .../test_readers/test_duckdb/test_ddb_xml.py | 84 ++----- .../test_readers/test_spark/test_spark_csv.py | 62 ++--- .../test_spark/test_spark_json.py | 53 ++-- .../test_readers/test_utilities.py | 10 +- 20 files changed, 504 insertions(+), 272 deletions(-) create mode 100644 tests/test_core_engine/test_backends/test_readers/fixtures.py diff --git a/src/dve/core_engine/backends/base/reader.py b/src/dve/core_engine/backends/base/reader.py index 93b97071..db568711 100644 --- a/src/dve/core_engine/backends/base/reader.py +++ b/src/dve/core_engine/backends/base/reader.py @@ -1,9 +1,8 @@ """Abstract implementation of the file parser.""" from abc import ABC, abstractmethod -from collections.abc import Iterator from inspect import ismethod -from typing import Any, ClassVar, Optional, TypeVar +from typing import Any, Callable, ClassVar, Iterator, Optional, TypeVar from pydantic import BaseModel from typing_extensions import Protocol @@ -18,6 +17,7 @@ AllowedAdditionalReaderChecks, _ReaderAdditionalChecksConfig, ) +from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME from dve.core_engine.message import FeedbackMessage from dve.core_engine.type_hints import URI, ArbitraryFunction, WrapDecorator from dve.parser.file_handling.service import open_stream @@ -57,11 +57,11 @@ def reader_impl_decorator(func: ArbitraryFunction) -> ArbitraryFunction: return reader_impl_decorator - class BaseFileReader(ABC): """An abstract representation of a reader for some file type.""" __read_methods__: ClassVar[_ReadFunctions] = {} + """ A dictionary mapping implemented entity types to their read functions. @@ -91,10 +91,9 @@ class variable for the subclass. continue entity_type: Optional[type] = getattr(method, _ENTITY_TYPE_ATTR_NAME, None) - if entity_type is None: + if not entity_type: continue - - cls.__read_methods__[entity_type] = method # type: ignore + cls.__read_methods__[entity_type] = method # type: ignore @abstractmethod def read_to_py_iterator( @@ -137,9 +136,9 @@ def read_to_entity_type( self.raise_if_not_sensible_file(resource, entity_name) if entity_type == Iterator[dict[str, Any]]: - entity = self.read_to_py_iterator( + entity = self.filter_null_records_py_iterator(self.read_to_py_iterator( resource, entity_name, schema, all_model_fields # type: ignore - ) + )) else: @@ -148,13 +147,15 @@ def read_to_entity_type( except KeyError as err: raise ReaderLacksEntityTypeSupport(entity_type=entity_type) from err - entity = reader_func( + entity = self.filter_null_records( + reader_func( self, resource, entity_name, schema, all_model_fields=all_model_fields, # type: ignore ) + ) if config := additional_checks.get("check_empty"): if self.check_entity_empty(entity): @@ -235,3 +236,14 @@ def raise_if_not_sensible_file( error_message=self.ft_error_message, ), ) + + def filter_null_records_py_iterator(self, + records: Iterator[dict[str, Any]]) -> Iterator[dict[str, Any]]: + """Strip null records from py iterator""" + def _is_non_null_record(record: dict[str, Any]) -> bool: + return any(v is not None for k, v in record.items() if not k == RECORD_INDEX_COLUMN_NAME) + yield from filter(_is_non_null_record, records) + + def filter_null_records(self, entity: EntityType) -> EntityType: + """Strip null records from entity""" + raise NotImplementedError() diff --git a/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py b/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py index 55db0945..e3e14492 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py +++ b/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py @@ -17,12 +17,13 @@ from duckdb import DuckDBPyConnection, DuckDBPyRelation, StarExpression from duckdb.typing import DuckDBPyType from pandas import DataFrame +import polars as pl from pydantic import BaseModel from typing_extensions import Annotated, get_args, get_origin, get_type_hints from dve.common.error_utils import get_feedback_errors_uri from dve.core_engine.backends.base.utilities import _get_non_heterogenous_type -from dve.core_engine.backends.utilities import DEFAULT_ISO_FORMATS, datetime_format_to_regex +from dve.core_engine.backends.utilities import DEFAULT_ISO_FORMATS, polars_filter_null_records, 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 @@ -500,3 +501,13 @@ def get_duckdb_cast_statement_from_annotation( 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}") + + +def _ddb_filter_null_records(self, entity: DuckDBPyRelation): + df = polars_filter_null_records(entity.pl()) + return self._connection.sql("SELECT * from df") + +def duckdb_filter_null_recs(cls): + """Add method to class to filter null records for duckdb relations""" + setattr(cls, "filter_null_records", _ddb_filter_null_records) + return cls diff --git a/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py b/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py index dedd66bd..ee2170e8 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py +++ b/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py @@ -23,9 +23,11 @@ ) from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import ( duckdb_check_entity_empty, + duckdb_filter_null_recs, duckdb_record_index, duckdb_write_parquet, get_duckdb_type_from_annotation, + polars_filter_null_records, relation_is_empty, ) from dve.core_engine.backends.implementations.duckdb.types import SQLType @@ -36,7 +38,7 @@ from dve.core_engine.type_hints import URI, EntityName from dve.parser.file_handling import get_content_length - +@duckdb_filter_null_recs @duckdb_check_entity_empty @duckdb_record_index @duckdb_write_parquet @@ -134,7 +136,7 @@ def read_to_relation( # pylint: disable=unused-argument return rel - +@duckdb_check_entity_empty @polars_record_index class PolarsToDuckDBCSVReader(DuckDBCSVReader): """ diff --git a/src/dve/core_engine/backends/implementations/duckdb/readers/json.py b/src/dve/core_engine/backends/implementations/duckdb/readers/json.py index 84b601de..0eeda5ff 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/readers/json.py +++ b/src/dve/core_engine/backends/implementations/duckdb/readers/json.py @@ -10,6 +10,7 @@ from dve.core_engine.backends.base.reader import BaseFileReader, read_function from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import ( + duckdb_filter_null_recs, duckdb_check_entity_empty, duckdb_record_index, duckdb_write_parquet, @@ -18,7 +19,7 @@ from dve.core_engine.backends.implementations.duckdb.types import SQLType from dve.core_engine.type_hints import URI, EntityName - +@duckdb_filter_null_recs @duckdb_check_entity_empty @duckdb_record_index @duckdb_write_parquet diff --git a/src/dve/core_engine/backends/implementations/duckdb/readers/xml.py b/src/dve/core_engine/backends/implementations/duckdb/readers/xml.py index 7e591e58..5a6d6052 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/readers/xml.py +++ b/src/dve/core_engine/backends/implementations/duckdb/readers/xml.py @@ -12,6 +12,7 @@ from dve.core_engine.backends.exceptions import CriticalMessageBearingError from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import ( duckdb_check_entity_empty, + duckdb_filter_null_recs, duckdb_write_parquet, ) from dve.core_engine.backends.readers.xml import XMLStreamReader @@ -22,7 +23,7 @@ ) from dve.core_engine.type_hints import URI - +@duckdb_filter_null_recs @duckdb_check_entity_empty @polars_record_index @duckdb_write_parquet diff --git a/src/dve/core_engine/backends/implementations/duckdb/utilities.py b/src/dve/core_engine/backends/implementations/duckdb/utilities.py index 62115007..6e2e5021 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/utilities.py +++ b/src/dve/core_engine/backends/implementations/duckdb/utilities.py @@ -38,4 +38,4 @@ def multiexpr_string_to_columns(expressions: str) -> list[str]: column expressions """ expression_list = _split_multiexpr_string(expressions) - return expr_array_to_columns(expression_list) + return expr_array_to_columns(expression_list) diff --git a/src/dve/core_engine/backends/implementations/spark/readers/csv.py b/src/dve/core_engine/backends/implementations/spark/readers/csv.py index 6b79bb35..cdda7f07 100644 --- a/src/dve/core_engine/backends/implementations/spark/readers/csv.py +++ b/src/dve/core_engine/backends/implementations/spark/readers/csv.py @@ -13,6 +13,7 @@ from dve.core_engine.backends.implementations.spark.spark_helpers import ( get_type_from_annotation, spark_check_entity_empty, + spark_filter_null_recs, spark_record_index, spark_write_parquet, ) @@ -20,7 +21,7 @@ from dve.core_engine.type_hints import URI, EntityName from dve.parser.file_handling import get_content_length - +@spark_filter_null_recs @spark_check_entity_empty @spark_record_index @spark_write_parquet diff --git a/src/dve/core_engine/backends/implementations/spark/readers/json.py b/src/dve/core_engine/backends/implementations/spark/readers/json.py index 6404231c..29268042 100644 --- a/src/dve/core_engine/backends/implementations/spark/readers/json.py +++ b/src/dve/core_engine/backends/implementations/spark/readers/json.py @@ -12,6 +12,7 @@ from dve.core_engine.backends.implementations.spark.spark_helpers import ( get_type_from_annotation, spark_check_entity_empty, + spark_filter_null_recs, spark_record_index, spark_write_parquet, ) @@ -19,6 +20,7 @@ from dve.parser.file_handling import get_content_length +@spark_filter_null_recs @spark_check_entity_empty @spark_record_index @spark_write_parquet diff --git a/src/dve/core_engine/backends/implementations/spark/readers/xml.py b/src/dve/core_engine/backends/implementations/spark/readers/xml.py index 275182e8..c08b4f60 100644 --- a/src/dve/core_engine/backends/implementations/spark/readers/xml.py +++ b/src/dve/core_engine/backends/implementations/spark/readers/xml.py @@ -18,6 +18,7 @@ df_is_empty, get_type_from_annotation, spark_check_entity_empty, + spark_filter_null_recs, spark_record_index, spark_write_parquet, ) @@ -29,7 +30,7 @@ SparkXMLMode = Literal["PERMISSIVE", "FAILFAST", "DROPMALFORMED"] """The mode to use when parsing XML files with Spark.""" - +@spark_filter_null_recs @spark_check_entity_empty @spark_record_index @spark_write_parquet @@ -56,7 +57,7 @@ def read_to_dataframe( ) ) - +@spark_filter_null_recs @spark_check_entity_empty @spark_record_index @spark_write_parquet diff --git a/src/dve/core_engine/backends/implementations/spark/spark_helpers.py b/src/dve/core_engine/backends/implementations/spark/spark_helpers.py index 1257306e..f6083fa7 100644 --- a/src/dve/core_engine/backends/implementations/spark/spark_helpers.py +++ b/src/dve/core_engine/backends/implementations/spark/spark_helpers.py @@ -638,3 +638,10 @@ def get_spark_cast_statement_from_annotation( 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}") + +def _spark_filter_null_records(self, entity: DataFrame) -> DataFrame: + return entity.dropna(how="all", subset=[cl for cl in entity.columns if not cl == RECORD_INDEX_COLUMN_NAME]) + +def spark_filter_null_recs(cls): + setattr(cls, "filter_null_records", _spark_filter_null_records) + return cls diff --git a/src/dve/core_engine/backends/utilities.py b/src/dve/core_engine/backends/utilities.py index af2f7573..01393221 100644 --- a/src/dve/core_engine/backends/utilities.py +++ b/src/dve/core_engine/backends/utilities.py @@ -272,3 +272,10 @@ def polars_record_index(cls): setattr(cls, "add_record_index", _add_polars_record_index) setattr(cls, "drop_record_index", _drop_polars_record_index) return cls + +def polars_filter_null_records(entity: pl.DataFrame) -> pl.DataFrame: + return entity.filter( + ~pl.all_horizontal(pl.col([cl for cl in entity.columns + if not cl == RECORD_INDEX_COLUMN_NAME]).is_null() + ) + ) \ No newline at end of file diff --git a/tests/test_core_engine/test_backends/fixtures.py b/tests/test_core_engine/test_backends/fixtures.py index 31c23d7d..9b34c943 100644 --- a/tests/test_core_engine/test_backends/fixtures.py +++ b/tests/test_core_engine/test_backends/fixtures.py @@ -589,4 +589,28 @@ def nested_parquet_custom_dc_err_details(temp_dir): yield file_path +@pytest.fixture +def temp_json_file_w_null_recs(temp_dir: Path): + + class SimpleModel(BaseModel): + varchar_field: str + bigint_field: int + date_field: date + timestamp_field: datetime + + field_names: list[str] = ["varchar_field","bigint_field","date_field","timestamp_field"] + typed_data = [ + ["hi", 1, date(2023, 1, 3), datetime(2023, 1, 3, 12, 0, 3)], + [None, None, None, None], + ["bye", 3, date(2023, 3, 7), datetime(2023, 5, 9, 15, 21, 53)], + [None, None, None, None] + ] + + test_data = [dict(zip(field_names, rw)) for rw in typed_data] + + with open(temp_dir.joinpath("test.json"), mode="w") as json_file: + json.dump(test_data, json_file, default=str) + + yield temp_dir.joinpath("test.json"), test_data, SimpleModel + diff --git a/tests/test_core_engine/test_backends/test_readers/fixtures.py b/tests/test_core_engine/test_backends/test_readers/fixtures.py new file mode 100644 index 00000000..4451553b --- /dev/null +++ b/tests/test_core_engine/test_backends/test_readers/fixtures.py @@ -0,0 +1,230 @@ + + +from datetime import date, datetime +import json +from pathlib import Path +from tempfile import TemporaryDirectory +from typing import Any, Iterator +from pydantic import BaseModel +import pytest +import polars as pl +from lxml import etree as ET + +## Models +class SimpleModel(BaseModel): + id_field: int + varchar_field: str + bigint_field: int + date_field: date + timestamp_field: datetime + + +class SimpleHeaderModel(BaseModel): + header_1: str + header_2: str + + +class VerySimpleModel(BaseModel): + test_col: str + +@pytest.fixture +def temp_dir(): + with TemporaryDirectory(prefix="reader_testing") as temp_dir: + yield Path(temp_dir) + +## CSV + +@pytest.fixture(scope="function") +def temp_csv_file(temp_dir: Path): + header: str = "id_field,varchar_field,bigint_field,date_field,timestamp_field" + typed_data = [ + [1, "hi", 1, date(2023, 1, 3), datetime(2023, 1, 3, 12, 0, 3)], + [2, "bye", 2, date(2023, 3, 7), datetime(2023, 5, 9, 15, 21, 53)], + ] + + with open(temp_dir.joinpath("dummy.csv"), mode="w") as csv_file: + csv_file.write(header + "\n") + for rw in typed_data: + csv_file.write(",".join([str(val) for val in rw]) + "\n") + + yield temp_dir.joinpath("dummy.csv"), header, typed_data, SimpleModel + + +@pytest.fixture(scope="function") +def temp_csv_file_additional_fields(temp_dir: Path) -> Iterator[str]: + test_df = pl.DataFrame({"test_col": ["fine"], "test_col2": ["wow"]}) + file_uri = temp_dir.joinpath("test_additional_fields.csv").as_posix() + test_df.write_csv( + file_uri, + include_header=True, + quote_style="always" + ) + + yield file_uri + + +@pytest.fixture(scope="function") +def temp_csv_file_missing_fields(temp_dir: Path) -> Iterator[str]: + test_df = pl.DataFrame({"header_1": ["fine"]}) + file_uri = temp_dir.joinpath("test_missing_fields.csv").as_posix() + test_df.write_csv( + file_uri, + include_header=True, + quote_style="always" + ) + + yield file_uri + + +@pytest.fixture +def temp_empty_csv_file(temp_dir: Path): + with open(temp_dir.joinpath("empty.csv"), mode="w"): + pass + + yield temp_dir.joinpath("empty.csv"), SimpleModel + +@pytest.fixture +def temp_csv_with_null_strings(temp_dir: Path): + test_df = pl.DataFrame({"test_col": ["fine", " ", " "]}) + file_uri = temp_dir.joinpath("test_empty_string1.csv").as_posix() + test_df.write_csv( + file_uri, + include_header=True, + quote_style="always" + ) + yield file_uri, VerySimpleModel + +@pytest.fixture +def temp_csv_with_null_records(temp_dir: Path): + data = {"id_field": ["1", "2" ,"", ""], + "varchar_field": ["fine", " ", " ", " "], + "bigint_field": ["", "3", "", " "], + "date_field": [date(2023, 1, 3),date(2023, 1, 5), None, None], + "timestamp_field": [None, datetime(2023, 1, 3, 12, 0, 3), None, None]} + test_df = pl.DataFrame(data) + file_uri = temp_dir.joinpath("test_remove_null_recs1.csv").as_posix() + test_df.write_csv( + file_uri, + include_header=True, + quote_style="always" + ) + yield file_uri, SimpleModel + +## JSON + +@pytest.fixture +def temp_json_file(temp_dir: Path): + field_names: list[str] = ["id_field","varchar_field","bigint_field","date_field","timestamp_field"] + typed_data = [ + [1,"hi", 1, date(2023, 1, 3), datetime(2023, 1, 3, 12, 0, 3)], + [2,"bye", 2, date(2023, 3, 7), datetime(2023, 5, 9, 15, 21, 53)], + ] + + test_data = [dict(zip(field_names, rw)) for rw in typed_data] + + with open(temp_dir.joinpath("test.json"), mode="w") as json_file: + json.dump(test_data, json_file, default=str) + + yield temp_dir.joinpath("test.json"), test_data, SimpleModel + +@pytest.fixture +def temp_json_file_w_null_recs(temp_dir: Path): + field_names: list[str] = ["id_field","varchar_field","bigint_field","date_field","timestamp_field"] + typed_data = [ + [1,"hi", 1, date(2023, 1, 3), datetime(2023, 1, 3, 12, 0, 3)], + [None, None, None, None], + [3,"bye", 3, date(2023, 3, 7), datetime(2023, 5, 9, 15, 21, 53)], + [None, None, None, None] + ] + + test_data = [dict(zip(field_names, rw)) for rw in typed_data] + + with open(temp_dir.joinpath("test.json"), mode="w") as json_file: + json.dump(test_data, json_file, default=str) + + yield temp_dir.joinpath("test.json"), test_data, SimpleModel + +## XML + +@pytest.fixture +def temp_xml_file(temp_dir: Path): + header_data: list[dict[str, str]] = [{ + "school_name": "Meadow Fields", + "category": "Primary", + "headteacher": "Mrs Smith", + }] + class_data: list[dict[str, dict[str, str]]] = [{ + "year_1": {"class_size": "10", "teacher": "Mrs Armitage"}, + "year_2": {"class_size": "12", "teacher": "Mr Barney"}, + }] + + class HeaderModel(BaseModel): + school_name: str + category: str + headteacher: str + + class ClassInfo(BaseModel): + class_size: int + teacher: str + + class ClassDataModel(BaseModel): + year_1: ClassInfo + year_2: ClassInfo + + root = ET.Element("root") + header = ET.SubElement(root, "Header") + for nm, val in header_data[0].items(): + _tag = ET.SubElement(header, nm) + _tag.text = val + + for dta in class_data: + data = ET.SubElement(root, "ClassData") + for nm, val in dta.items(): + _parent_tag = ET.SubElement(data, nm) + for sub_nm, sub_val in val.items(): + _child_tag = ET.SubElement(_parent_tag, sub_nm) + _child_tag.text = sub_val + + with open(temp_dir.joinpath("test.xml"), mode="wb") as xml_fle: + xml_fle.write(ET.tostring(root)) + + yield temp_dir.joinpath("test.xml"), HeaderModel, header_data, ClassDataModel, class_data + +@pytest.fixture +def temp_xml_file_w_null_recs(temp_dir: Path): + header_data: list[dict[str, str]] = [{ + "school_name": "Meadow Fields", + "category": "Primary", + "headteacher": "Mrs Smith", + }] + class_data: list[ dict[str, Any]] = [ + {"year_group": 1, "class_size": "10", "teacher": "Mrs Armitage"}, + {"year_group": 2, "class_size": "12", "teacher": "Mr Barney"}, + {"year_group": None, "class_size": None, "teacher": None}] + + class HeaderModel(BaseModel): + school_name: str + category: str + headteacher: str + + class ClassInfo(BaseModel): + year_group: int + class_size: int + teacher: str + + root = ET.Element("root") + header = ET.SubElement(root, "Header") + for nm, val in header_data[0].items(): + _tag = ET.SubElement(header, nm) + _tag.text = val + + for dta in class_data: + data = ET.SubElement(root, "ClassData") + for nm, val in dta.items(): + _child_tag = ET.SubElement(data, nm) + _child_tag.text = str(val) if val else None + + with open(temp_dir.joinpath("test_with_nulls.xml"), mode="wb") as xml_fle: + xml_fle.write(ET.tostring(root)) + + yield temp_dir.joinpath("test_with_nulls.xml"), HeaderModel, header_data, ClassInfo, class_data \ No newline at end of file diff --git a/tests/test_core_engine/test_backends/test_readers/test_csv.py b/tests/test_core_engine/test_backends/test_readers/test_csv.py index 413b6145..68f8e470 100644 --- a/tests/test_core_engine/test_backends/test_readers/test_csv.py +++ b/tests/test_core_engine/test_backends/test_readers/test_csv.py @@ -4,7 +4,7 @@ # pylint: disable=missing-class-docstring import csv from pathlib import Path -from typing import Dict, Iterator, Optional +from typing import Any, Dict, Iterator, Optional from uuid import uuid4 import pandas as pd @@ -21,6 +21,7 @@ from dve.core_engine.backends.readers import CSVFileReader from dve.core_engine.backends.readers.utilities import get_all_model_fields from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME +from dve.core_engine.type_hints import URI from ....conftest import get_test_file_path from ....fixtures import temp_dir @@ -57,6 +58,19 @@ def null_values_location(temp_dir: str) -> Iterator[str]: yield path.as_uri() +@pytest.fixture +def csv_data_with_null_records(temp_dir: str) -> Iterator[tuple[URI, BaseModel]]: + path = Path(temp_dir).joinpath("test_with_null_recs.csv") + header = [fld for fld in PlanetsSubset.model_fields] + data = [("a_planet", " ", ""), ("", "", ""), ("", "3.1",""), ("", "", "1.9")] + with open(path, mode="w") as fle: + fle.write(",".join(header) + "\n") + for rw in data: + fle.write(",".join(rw) + "\n") + + yield path.as_posix(), PlanetsSubset + + @pytest.fixture(scope="function") def pipe_delimited_location(temp_dir: str) -> Iterator[str]: """The URI of a file containing pipe-delimited values.""" @@ -318,3 +332,15 @@ def test_base_csv_reader_with_missing_fields( error_msg = exc_info.value.message assert "additional_fields" not in error_msg.record["test"] assert error_msg.record["test"] == "missing fields: random_null;" + + def test_base_csv_reader_remove_null_recs( + self, + csv_data_with_null_records + ): + uri, mdl = csv_data_with_null_records + reader = CSVFileReader() + + result = list(reader.read_to_entity_type(entity_type=Iterator[dict[str, Any]], resource=uri, entity_name = "test", schema=mdl)) + + assert len(result) == 3 + diff --git a/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_csv.py b/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_csv.py index 8bc97754..c8d87c90 100644 --- a/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_csv.py +++ b/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_csv.py @@ -25,82 +25,21 @@ ) from dve.core_engine.backends.utilities import stringify_model from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME +from tests.test_core_engine.test_backends.test_readers.fixtures import ( + SimpleHeaderModel, + SimpleModel, + VerySimpleModel, + temp_dir, + temp_csv_file, + temp_csv_file_additional_fields, + temp_csv_file_missing_fields, + temp_empty_csv_file, + temp_csv_with_null_strings, + temp_csv_with_null_records) # pylint: disable=C0103,C0115,C0116,W0621 -class SimpleModel(BaseModel): - varchar_field: str - bigint_field: int - date_field: date - timestamp_field: datetime - - -class SimpleHeaderModel(BaseModel): - header_1: str - header_2: str - - -class VerySimpleModel(BaseModel): - test_col: str - - -@pytest.fixture -def temp_dir(): - with TemporaryDirectory(prefix="ddb_test_csv_reader") as temp_dir: - yield Path(temp_dir) - - -@pytest.fixture(scope="function") -def temp_csv_file(temp_dir: Path): - header: str = "varchar_field,bigint_field,date_field,timestamp_field" - typed_data = [ - ["hi", 1, date(2023, 1, 3), datetime(2023, 1, 3, 12, 0, 3)], - ["bye", 2, date(2023, 3, 7), datetime(2023, 5, 9, 15, 21, 53)], - ] - - with open(temp_dir.joinpath("dummy.csv"), mode="w") as csv_file: - csv_file.write(header + "\n") - for rw in typed_data: - csv_file.write(",".join([str(val) for val in rw]) + "\n") - - yield temp_dir.joinpath("dummy.csv"), header, typed_data, SimpleModel - - -@pytest.fixture(scope="function") -def temp_csv_file_additional_fields(temp_dir: Path) -> Iterator[str]: - test_df = pl.DataFrame({"test_col": ["fine"], "test_col2": ["wow"]}) - file_uri = temp_dir.joinpath("test_additional_fields.csv").as_posix() - test_df.write_csv( - file_uri, - include_header=True, - quote_style="always" - ) - - yield file_uri - - -@pytest.fixture(scope="function") -def temp_csv_file_missing_fields(temp_dir: Path) -> Iterator[str]: - test_df = pl.DataFrame({"header_1": ["fine"]}) - file_uri = temp_dir.joinpath("test_missing_fields.csv").as_posix() - test_df.write_csv( - file_uri, - include_header=True, - quote_style="always" - ) - - yield file_uri - - -@pytest.fixture -def temp_empty_csv_file(temp_dir: Path): - with open(temp_dir.joinpath("empty.csv"), mode="w"): - pass - - yield temp_dir.joinpath("empty.csv"), SimpleModel - - class TestDuckDBCSVReader: """Test DuckDBCSVReader""" @@ -147,15 +86,8 @@ def test_ddb_csv_read_empty_file(self, temp_empty_csv_file): with pytest.raises(EmptyFileError): reader.read_to_relation(str(uri), "test", mdl) - def test_DuckDBCSVReader_with_null_empty_strings(self, temp_dir): - test_df = pl.DataFrame({"test_col": ["fine", " ", " "]}) - file_uri = temp_dir.joinpath("test_empty_string1.csv").as_posix() - test_df.write_csv( - file_uri, - include_header=True, - quote_style="always" - ) - + def test_DuckDBCSVReader_with_null_empty_strings(self, temp_csv_with_null_strings): + uri, mdl = temp_csv_with_null_strings reader = DuckDBCSVReader( header=True, delim=",", @@ -164,14 +96,33 @@ def test_DuckDBCSVReader_with_null_empty_strings(self, temp_dir): null_empty_strings=True, ) - entity = reader.read_to_relation(file_uri, "test", VerySimpleModel) + entity = reader.read_to_relation(uri, "test", mdl) assert entity.shape[0] == 3 assert entity.filter("test_col IS NULL").shape[0] == 2 + + def test_DuckDBCSVReader_removes_null_records(self, temp_csv_with_null_records): + uri, mdl = temp_csv_with_null_records + + + reader = DuckDBCSVReader( + header=True, + delim=",", + quotechar='"', + connection=duckdb.connect(), + null_empty_strings=True, + ) + + entity = reader.read_to_entity_type(DuckDBPyRelation, uri, "test", stringify_model(mdl)) + + assert entity.shape[0] == 2 + + assert entity.select("id_field").pl().to_dict(as_series=False).get("id_field") == ["1" , "2"] + def test_DuckDBCSVReader_with_malformed_header(self, temp_dir): - test_data_headers = '"varchar_field,bigint_field,date_field,timestamp_field"' - row_data = "hello,1,2023-04-01,2023-04-01T12:30:00" + test_data_headers = '"id_field,varchar_field,bigint_field,date_field,timestamp_field"' + row_data = "1,hello,1,2023-04-01,2023-04-01T12:30:00" temp_id = uuid4().hex fqp = Path(temp_dir, f"{temp_id}.csv") @@ -222,6 +173,22 @@ def test_PolarsToDuckDBCSVReader_with_null_empty_strings(self, temp_dir): assert entity.shape[0] == 3 assert entity.filter("test_col IS NULL").shape[0] == 2 + + def test_PolarsToDuckDBCSVReader_removes_null_records(self, temp_csv_with_null_records): + uri, mdl = temp_csv_with_null_records + + reader = PolarsToDuckDBCSVReader( + header=True, + delim=",", + quotechar='"', + connection=duckdb.connect(), + null_empty_strings=True, + ) + + entity = reader.read_to_entity_type(DuckDBPyRelation, uri, "test", stringify_model(mdl)) + + assert entity.shape[0] == 2 + assert entity.select("id_field").pl().to_dict(as_series=False).get("id_field") == ["1", "2"] def test_PolarsToDuckDBCSVReader_with_malformed_header(self, temp_dir): test_data_headers = '"varchar_field,bigint_field,date_field,timestamp_field"' @@ -310,3 +277,26 @@ def test_DuckDBCSVRepeatingHeaderReader_with_null_empty_strings(self, temp_dir): assert entity.shape[0] == 1 assert entity.filter("header_2 IS NULL").shape[0] == 1 + + def test_DuckDBCSVRepeatingHeaderReader_removes_null_records(self, temp_dir): + test_df = pl.DataFrame({"header_1": [""], + "header_2": [""]}) + file_uri = temp_dir.joinpath("test_remove_null_header.csv").as_posix() + test_df.write_csv( + file_uri, + include_header=True, + quote_style="always" + ) + + reader = DuckDBCSVRepeatingHeaderReader( + header=True, + delim=",", + quotechar='"', + connection=duckdb.connect(), + null_empty_strings=True, + ) + + entity = reader.read_to_entity_type(DuckDBPyRelation, file_uri, "test", SimpleHeaderModel) + + assert entity.shape[0] == 0 + diff --git a/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_json.py b/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_json.py index ae8e4adc..b7dfe859 100644 --- a/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_json.py +++ b/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_json.py @@ -16,43 +16,14 @@ from dve.core_engine.backends.utilities import stringify_model from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME from tests.test_core_engine.test_backends.fixtures import duckdb_connection +from tests.test_core_engine.test_backends.test_readers.fixtures import ( + temp_dir, + temp_json_file, + temp_json_file_w_null_recs) -class SimpleModel(BaseModel): - varchar_field: str - bigint_field: int - date_field: date - timestamp_field: datetime -@pytest.fixture -def temp_dir(): - with TemporaryDirectory(prefix="ddb_test_json_reader") as temp_dir: - yield Path(temp_dir) - - -@pytest.fixture -def temp_json_file(temp_dir: Path): - field_names: List[str] = ["varchar_field","bigint_field","date_field","timestamp_field"] - typed_data = [ - ["hi", 1, date(2023, 1, 3), datetime(2023, 1, 3, 12, 0, 3)], - ["bye", 2, date(2023, 3, 7), datetime(2023, 5, 9, 15, 21, 53)], - ] - - test_data = [dict(zip(field_names, rw)) for rw in typed_data] - - with open(temp_dir.joinpath("test.json"), mode="w") as json_file: - json.dump(test_data, json_file, default=str) - - yield temp_dir.joinpath("test.json"), test_data, SimpleModel - - -class SimpleModel(BaseModel): - varchar_field: str - bigint_field: int - date_field: date - timestamp_field: datetime - def test_ddb_json_reader_all_str(temp_json_file): uri, data, mdl = temp_json_file @@ -104,3 +75,12 @@ def test_ddb_json_write_parquet_py_iterator(temp_json_file): parquet_data = sorted(conn.read_parquet(target_loc).pl().iter_rows(named=True), key= lambda x: x.get("bigint_field")) assert parquet_data == list(data) + +def test_ddb_json_remove_null_records(temp_json_file_w_null_recs): + uri, _, mdl = temp_json_file_w_null_recs + reader = DuckDBJSONReader() + rel: DuckDBPyRelation = reader.read_to_entity_type( + DuckDBPyRelation, uri.as_posix(), "test", stringify_model(mdl) + ) + assert rel.shape[0] == 2 + assert rel.select("id_field").pl().to_dict(as_series=False).get("id_field") == ["1", "3"] diff --git a/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_xml.py b/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_xml.py index 1c12a7b8..0341bb6e 100644 --- a/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_xml.py +++ b/tests/test_core_engine/test_backends/test_readers/test_duckdb/test_ddb_xml.py @@ -1,68 +1,15 @@ -from datetime import date, datetime -from pathlib import Path -from tempfile import TemporaryDirectory -from typing import Dict, List +from pathlib import Path import duckdb -import pytest from duckdb import DuckDBPyRelation -from lxml import etree as ET -from pydantic import BaseModel from dve.core_engine.backends.implementations.duckdb.readers.xml import DuckDBXMLStreamReader from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME - - -@pytest.fixture -def temp_dir(): - with TemporaryDirectory(prefix="ddb_test_xml_reader") as temp_dir: - yield Path(temp_dir) - - -@pytest.fixture -def temp_xml_file(temp_dir: Path): - header_data: list[dict[str, str]] = [{ - "school_name": "Meadow Fields", - "category": "Primary", - "headteacher": "Mrs Smith", - }] - class_data: list[dict[str, Dict[str, str]]] = [{ - "year_1": {"class_size": "10", "teacher": "Mrs Armitage"}, - "year_2": {"class_size": "12", "teacher": "Mr Barney"}, - }] - - class HeaderModel(BaseModel): - school_name: str - category: str - headteacher: str - - class ClassInfo(BaseModel): - class_size: int - teacher: str - - class ClassDataModel(BaseModel): - year_1: ClassInfo - year_2: ClassInfo - - root = ET.Element("root") - header = ET.SubElement(root, "Header") - for nm, val in header_data[0].items(): - _tag = ET.SubElement(header, nm) - _tag.text = val - - for dta in class_data: - data = ET.SubElement(root, "ClassData") - for nm, val in dta.items(): - _parent_tag = ET.SubElement(data, nm) - for sub_nm, sub_val in val.items(): - _child_tag = ET.SubElement(_parent_tag, sub_nm) - _child_tag.text = sub_val - - with open(temp_dir.joinpath("test.xml"), mode="wb") as xml_fle: - xml_fle.write(ET.tostring(root)) - - yield temp_dir.joinpath("test.xml"), HeaderModel, header_data, ClassDataModel, class_data - +from tests.test_core_engine.test_backends.test_readers.fixtures import ( + temp_dir, + temp_xml_file, + temp_xml_file_w_null_recs +) def test_ddb_xml_reader_all_str(temp_xml_file): uri, header_model, header_data, class_data_model, class_data = temp_xml_file @@ -116,3 +63,22 @@ def test_ddb_xml_reader_write_parquet(temp_xml_file): assert class_parquet_rel.df().to_dict(orient="records") == class_rel.df().to_dict( orient="records" ) + +def test_ddb_xml_reader_remove_null_recs(temp_xml_file_w_null_recs): + uri, header_model, _, class_data_model, _ = temp_xml_file_w_null_recs + ddb_conn = duckdb.connect() + header_reader = DuckDBXMLStreamReader( + connection=ddb_conn, root_tag="root", record_tag="Header" + ) + class_reader = DuckDBXMLStreamReader( + connection=ddb_conn, root_tag="root", record_tag="ClassData" + ) + header_rel: DuckDBPyRelation = header_reader.read_to_entity_type( + DuckDBPyRelation, uri.as_uri(), "header", header_model + ) + class_rel: DuckDBPyRelation = class_reader.read_to_entity_type( + DuckDBPyRelation, uri.as_uri(), "class_data", class_data_model + ) + assert header_rel.count("*").fetchone()[0] == 1 + assert class_rel.count("*").fetchone()[0] == 2 + assert class_rel.select("year_group").pl().to_dict(as_series=False).get("year_group") == ["1", "2"] diff --git a/tests/test_core_engine/test_backends/test_readers/test_spark/test_spark_csv.py b/tests/test_core_engine/test_backends/test_readers/test_spark/test_spark_csv.py index f28f0ac8..9f2fbbd1 100644 --- a/tests/test_core_engine/test_backends/test_readers/test_spark/test_spark_csv.py +++ b/tests/test_core_engine/test_backends/test_readers/test_spark/test_spark_csv.py @@ -5,51 +5,20 @@ # pylint: disable=C0103 # pylint: disable=C0115 -import tempfile -from pathlib import Path - -import polars as pl -import pytest -from pydantic import BaseModel from pyspark.sql import DataFrame, Row, SparkSession from pyspark.sql.types import StringType, StructField, StructType -from dve.core_engine.backends.exceptions import MessageBearingError from dve.core_engine.backends.implementations.spark.readers.csv import SparkCSVReader +from dve.core_engine.backends.utilities import stringify_model +from tests.test_core_engine.test_backends.test_readers.fixtures import ( + temp_dir, + temp_csv_with_null_strings, + temp_csv_with_null_records +) -class SparkCSVTestModel(BaseModel): - test_col: str - - -class SparkCSVTestModelAdditionalField(SparkCSVTestModel): - test_col2: str - - -@pytest.fixture(scope="function") -def spark_null_csv_resource(): - test_df = pl.DataFrame({"test_col": ["fine", " ", " "]}) - - with tempfile.TemporaryDirectory() as tdir: - resource_uri = Path(tdir, "test_spark_csv_reader.csv").as_posix() - test_df.write_csv(resource_uri, include_header=True, quote_style="always") - - yield resource_uri - - -@pytest.fixture(scope="function") -def spark_additional_fields(): - test_df = pl.DataFrame({"test_col": ["abc"], "test_col2": ["def"]}) - - with tempfile.TemporaryDirectory() as tdir: - resource_uri = Path(tdir, "test_spark_csv_reader_add_fields.csv").as_posix() - test_df.write_csv(resource_uri, include_header=True, quote_style="always") - - yield resource_uri - - -def test_SparkCSVReader_clean_empty_strings(spark: SparkSession, spark_null_csv_resource): - resource_uri = spark_null_csv_resource +def test_SparkCSVReader_clean_empty_strings(spark: SparkSession, temp_csv_with_null_strings): + resource_uri, mdl = temp_csv_with_null_strings expected_df = spark.createDataFrame( [ Row( @@ -66,7 +35,20 @@ def test_SparkCSVReader_clean_empty_strings(spark: SparkSession, spark_null_csv_ reader = SparkCSVReader(null_empty_strings=True, spark_session=spark) result_df: DataFrame = reader.read_to_dataframe( - resource=resource_uri, entity_name="test", schema=SparkCSVTestModel + resource=resource_uri, entity_name="test", schema=stringify_model(mdl) ) assert result_df.exceptAll(expected_df).count() == 0 + +def test_SparkCSVReader_remove_null_records(spark, temp_csv_with_null_records): + uri, mdl = temp_csv_with_null_records + + reader = SparkCSVReader(null_empty_strings=True, spark_session=spark) + + result_df: DataFrame = reader.read_to_entity_type( + entity_type=DataFrame, resource=uri, entity_name="test", schema=stringify_model(mdl) + ) + + assert result_df.count() == 2 + assert [rw.id_field for rw in result_df.select("id_field").collect()] == ["1", "2"] + diff --git a/tests/test_core_engine/test_backends/test_readers/test_spark/test_spark_json.py b/tests/test_core_engine/test_backends/test_readers/test_spark/test_spark_json.py index 46d4729c..f5a5e91a 100644 --- a/tests/test_core_engine/test_backends/test_readers/test_spark/test_spark_json.py +++ b/tests/test_core_engine/test_backends/test_readers/test_spark/test_spark_json.py @@ -15,42 +15,11 @@ from dve.core_engine.backends.implementations.spark.readers.json import SparkJSONReader from dve.core_engine.backends.utilities import stringify_model from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME - - -class SimpleModel(BaseModel): - varchar_field: str - bigint_field: int - date_field: date - timestamp_field: datetime - - -@pytest.fixture -def temp_dir(): - with TemporaryDirectory(prefix="spark_test_json_reader") as temp_dir: - yield Path(temp_dir) - - -@pytest.fixture -def temp_json_file(temp_dir: Path): - field_names: List[str] = ["varchar_field","bigint_field","date_field","timestamp_field"] - typed_data = [ - ["hi", 1, date(2023, 1, 3), datetime(2023, 1, 3, 12, 0, 3)], - ["bye", 2, date(2023, 3, 7), datetime(2023, 5, 9, 15, 21, 53)], - ] - - test_data = [dict(zip(field_names, rw)) for rw in typed_data] - - with open(temp_dir.joinpath("test.json"), mode="w") as json_file: - json.dump(test_data, json_file, default=str) - - yield temp_dir.joinpath("test.json"), test_data, SimpleModel - - -class SimpleModel(BaseModel): - varchar_field: str - bigint_field: int - date_field: date - timestamp_field: datetime +from tests.test_core_engine.test_backends.test_readers.fixtures import ( + temp_dir, + temp_json_file, + temp_json_file_w_null_recs +) def test_spark_json_reader_all_str(temp_json_file): @@ -97,3 +66,15 @@ def test_spark_json_write_parquet_py_iterator(spark, temp_json_file): in spark.read.parquet(target_loc).collect()], key= lambda x: x.get("bigint_field")) assert parquet_data == list(data) + +def test_SparkJSONReader_remove_null_records(spark, temp_json_file_w_null_recs): + uri, _, mdl = temp_json_file_w_null_recs + + reader = SparkJSONReader() + + result_df: DataFrame = reader.read_to_entity_type( + entity_type=DataFrame, resource=uri.as_posix(), entity_name="test", schema=stringify_model(mdl) + ) + + assert result_df.count() == 2 + assert [rw.id_field for rw in result_df.select("id_field").collect()] == ["1", "3"] diff --git a/tests/test_core_engine/test_backends/test_readers/test_utilities.py b/tests/test_core_engine/test_backends/test_readers/test_utilities.py index 1e2be41b..6f181fa9 100644 --- a/tests/test_core_engine/test_backends/test_readers/test_utilities.py +++ b/tests/test_core_engine/test_backends/test_readers/test_utilities.py @@ -1,15 +1,23 @@ """Test utility functions & objects for readers""" +from datetime import date, datetime +import json +from pathlib import Path +from tempfile import TemporaryDirectory from pydantic import BaseModel +import pytest from dve.core_engine.backends.readers.utilities import get_all_model_fields +@pytest.fixture +def temp_dir(): + with TemporaryDirectory(prefix="ddb_test_json_reader") as temp_dir: + yield Path(temp_dir) class Model1(BaseModel): # pylint: disable=C0115 model1_field_1: str model1_field_2: int - class Model2(BaseModel): # pylint: disable=C0115 model2_field_1: str From b65b48832c5e469f6a1982565999adf0a54877eb Mon Sep 17 00:00:00 2001 From: stevenhsd <56357022+stevenhsd@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:50:21 +0100 Subject: [PATCH 2/2] style: address linting and sonarqube issues --- src/dve/core_engine/backends/base/reader.py | 44 +++++++++++-------- src/dve/core_engine/backends/exceptions.py | 5 ++- .../implementations/duckdb/duckdb_helpers.py | 10 +++-- .../implementations/duckdb/readers/csv.py | 12 +++-- .../implementations/duckdb/readers/json.py | 3 +- .../implementations/duckdb/readers/xml.py | 1 + .../implementations/duckdb/utilities.py | 2 +- .../implementations/spark/readers/csv.py | 1 + .../implementations/spark/readers/xml.py | 2 + .../implementations/spark/spark_helpers.py | 11 ++++- .../core_engine/backends/readers/utilities.py | 2 +- src/dve/core_engine/backends/utilities.py | 10 +++-- 12 files changed, 67 insertions(+), 36 deletions(-) diff --git a/src/dve/core_engine/backends/base/reader.py b/src/dve/core_engine/backends/base/reader.py index db568711..edec095f 100644 --- a/src/dve/core_engine/backends/base/reader.py +++ b/src/dve/core_engine/backends/base/reader.py @@ -2,7 +2,7 @@ from abc import ABC, abstractmethod from inspect import ismethod -from typing import Any, Callable, ClassVar, Iterator, Optional, TypeVar +from typing import Any, ClassVar, Iterator, Optional, TypeVar from pydantic import BaseModel from typing_extensions import Protocol @@ -10,7 +10,7 @@ from dve.core_engine.backends.exceptions import ( CriticalMessageBearingError, MessageBearingError, - ReaderLacksEntityTypeSupport + ReaderLacksEntityTypeSupport, ) from dve.core_engine.backends.types import EntityName, EntityType from dve.core_engine.configuration.v1 import ( @@ -57,11 +57,12 @@ def reader_impl_decorator(func: ArbitraryFunction) -> ArbitraryFunction: return reader_impl_decorator + class BaseFileReader(ABC): """An abstract representation of a reader for some file type.""" __read_methods__: ClassVar[_ReadFunctions] = {} - + """ A dictionary mapping implemented entity types to their read functions. @@ -91,9 +92,9 @@ class variable for the subclass. continue entity_type: Optional[type] = getattr(method, _ENTITY_TYPE_ATTR_NAME, None) - if not entity_type: + if entity_type is None: continue - cls.__read_methods__[entity_type] = method # type: ignore + cls.__read_methods__[entity_type] = method # type: ignore @abstractmethod def read_to_py_iterator( @@ -136,9 +137,11 @@ def read_to_entity_type( self.raise_if_not_sensible_file(resource, entity_name) if entity_type == Iterator[dict[str, Any]]: - entity = self.filter_null_records_py_iterator(self.read_to_py_iterator( - resource, entity_name, schema, all_model_fields # type: ignore - )) + entity = self.filter_null_records_py_iterator( + self.read_to_py_iterator( + resource, entity_name, schema, all_model_fields # type: ignore + ) + ) else: @@ -149,13 +152,13 @@ def read_to_entity_type( entity = self.filter_null_records( reader_func( - self, - resource, - entity_name, - schema, - all_model_fields=all_model_fields, # type: ignore - ) + self, + resource, + entity_name, + schema, + all_model_fields=all_model_fields, # type: ignore ) + ) if config := additional_checks.get("check_empty"): if self.check_entity_empty(entity): @@ -236,14 +239,17 @@ def raise_if_not_sensible_file( error_message=self.ft_error_message, ), ) - - def filter_null_records_py_iterator(self, - records: Iterator[dict[str, Any]]) -> Iterator[dict[str, Any]]: + + def filter_null_records_py_iterator( + self, records: Iterator[dict[str, Any]] + ) -> Iterator[dict[str, Any]]: """Strip null records from py iterator""" + def _is_non_null_record(record: dict[str, Any]) -> bool: - return any(v is not None for k, v in record.items() if not k == RECORD_INDEX_COLUMN_NAME) + return any(v is not None for k, v in record.items() if k != RECORD_INDEX_COLUMN_NAME) + yield from filter(_is_non_null_record, records) - + def filter_null_records(self, entity: EntityType) -> EntityType: """Strip null records from entity""" raise NotImplementedError() diff --git a/src/dve/core_engine/backends/exceptions.py b/src/dve/core_engine/backends/exceptions.py index f8d93079..99808c8a 100644 --- a/src/dve/core_engine/backends/exceptions.py +++ b/src/dve/core_engine/backends/exceptions.py @@ -32,6 +32,7 @@ def __init__(self, *args: object, messages: Messages) -> None: self.messages = messages """The messages to be returned as part of the error.""" + class CriticalMessageBearingError(BackendError): """ A backend error that comes with a pre-created message. @@ -44,6 +45,7 @@ def __init__(self, *args: object, message: FeedbackMessage) -> None: self.message = message """The message to be returned as part of the error.""" + class UnableToParseCSVError(CriticalMessageBearingError): """An error raised when unable to parse a CSV file""" @@ -60,7 +62,8 @@ def __init__( failure_type="submission", is_informational=False, error_type="csv read", - error_message=error_message or "Unable to parse the CSV file. Please check the structure of your CSV.", # pylint: disable=C0301 + error_message=error_message + or "Unable to parse the CSV file. Please check the structure of your CSV.", # pylint: disable=C0301 error_code=error_code or "MalformedCSV", ) ) diff --git a/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py b/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py index e3e14492..989512a4 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py +++ b/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py @@ -17,13 +17,16 @@ from duckdb import DuckDBPyConnection, DuckDBPyRelation, StarExpression from duckdb.typing import DuckDBPyType from pandas import DataFrame -import polars as pl from pydantic import BaseModel from typing_extensions import Annotated, get_args, get_origin, get_type_hints from dve.common.error_utils import get_feedback_errors_uri from dve.core_engine.backends.base.utilities import _get_non_heterogenous_type -from dve.core_engine.backends.utilities import DEFAULT_ISO_FORMATS, polars_filter_null_records, datetime_format_to_regex +from dve.core_engine.backends.utilities import ( + DEFAULT_ISO_FORMATS, + datetime_format_to_regex, + polars_filter_null_records, +) 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 @@ -504,9 +507,10 @@ def get_duckdb_cast_statement_from_annotation( def _ddb_filter_null_records(self, entity: DuckDBPyRelation): - df = polars_filter_null_records(entity.pl()) + df = polars_filter_null_records(entity.pl()) # pylint: disable=W0612 return self._connection.sql("SELECT * from df") + def duckdb_filter_null_recs(cls): """Add method to class to filter null records for duckdb relations""" setattr(cls, "filter_null_records", _ddb_filter_null_records) diff --git a/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py b/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py index ee2170e8..4a60dcb7 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py +++ b/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py @@ -27,7 +27,6 @@ duckdb_record_index, duckdb_write_parquet, get_duckdb_type_from_annotation, - polars_filter_null_records, relation_is_empty, ) from dve.core_engine.backends.implementations.duckdb.types import SQLType @@ -38,6 +37,7 @@ from dve.core_engine.type_hints import URI, EntityName from dve.parser.file_handling import get_content_length + @duckdb_filter_null_recs @duckdb_check_entity_empty @duckdb_record_index @@ -125,7 +125,8 @@ def read_to_relation( # pylint: disable=unused-argument raise UnableToParseCSVError( entity_name="csv_structure", error_code=self.ft_error_code, - error_message=self.ft_error_message or "Unable to parse CSV file. Structure is likely malformed.", # pylint: disable=C0301 + error_message=self.ft_error_message + or "Unable to parse CSV file. Structure is likely malformed.", # pylint: disable=C0301 ) from exc if self.null_empty_strings: @@ -136,6 +137,7 @@ def read_to_relation( # pylint: disable=unused-argument return rel + @duckdb_check_entity_empty @polars_record_index class PolarsToDuckDBCSVReader(DuckDBCSVReader): @@ -185,7 +187,8 @@ def read_to_relation( # pylint: disable=unused-argument raise UnableToParseCSVError( entity_name="csv_structure", error_code=self.ft_error_code, - error_message=self.ft_error_message or "Unable to parse CSV file. Structure is likely malformed.", # pylint: disable=C0301 + error_message=self.ft_error_message + or "Unable to parse CSV file. Structure is likely malformed.", # pylint: disable=C0301 ) from exc if self.null_empty_strings: @@ -202,7 +205,8 @@ def read_to_relation( # pylint: disable=unused-argument raise UnableToParseCSVError( entity_name="csv_structure", error_code=self.ft_error_code, - error_message=self.ft_error_message or "Found zero records after loading CSV. File is likely malformed.", # pylint: disable=C0301 + error_message=self.ft_error_message + or "Found zero records after loading CSV. File is likely malformed.", # pylint: disable=C0301 ) return entity diff --git a/src/dve/core_engine/backends/implementations/duckdb/readers/json.py b/src/dve/core_engine/backends/implementations/duckdb/readers/json.py index 0eeda5ff..a8ccf116 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/readers/json.py +++ b/src/dve/core_engine/backends/implementations/duckdb/readers/json.py @@ -10,8 +10,8 @@ from dve.core_engine.backends.base.reader import BaseFileReader, read_function from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import ( - duckdb_filter_null_recs, duckdb_check_entity_empty, + duckdb_filter_null_recs, duckdb_record_index, duckdb_write_parquet, get_duckdb_type_from_annotation, @@ -19,6 +19,7 @@ from dve.core_engine.backends.implementations.duckdb.types import SQLType from dve.core_engine.type_hints import URI, EntityName + @duckdb_filter_null_recs @duckdb_check_entity_empty @duckdb_record_index diff --git a/src/dve/core_engine/backends/implementations/duckdb/readers/xml.py b/src/dve/core_engine/backends/implementations/duckdb/readers/xml.py index 5a6d6052..7b753eef 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/readers/xml.py +++ b/src/dve/core_engine/backends/implementations/duckdb/readers/xml.py @@ -23,6 +23,7 @@ ) from dve.core_engine.type_hints import URI + @duckdb_filter_null_recs @duckdb_check_entity_empty @polars_record_index diff --git a/src/dve/core_engine/backends/implementations/duckdb/utilities.py b/src/dve/core_engine/backends/implementations/duckdb/utilities.py index 6e2e5021..62115007 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/utilities.py +++ b/src/dve/core_engine/backends/implementations/duckdb/utilities.py @@ -38,4 +38,4 @@ def multiexpr_string_to_columns(expressions: str) -> list[str]: column expressions """ expression_list = _split_multiexpr_string(expressions) - return expr_array_to_columns(expression_list) + return expr_array_to_columns(expression_list) diff --git a/src/dve/core_engine/backends/implementations/spark/readers/csv.py b/src/dve/core_engine/backends/implementations/spark/readers/csv.py index cdda7f07..440bf905 100644 --- a/src/dve/core_engine/backends/implementations/spark/readers/csv.py +++ b/src/dve/core_engine/backends/implementations/spark/readers/csv.py @@ -21,6 +21,7 @@ from dve.core_engine.type_hints import URI, EntityName from dve.parser.file_handling import get_content_length + @spark_filter_null_recs @spark_check_entity_empty @spark_record_index diff --git a/src/dve/core_engine/backends/implementations/spark/readers/xml.py b/src/dve/core_engine/backends/implementations/spark/readers/xml.py index c08b4f60..22cd9d71 100644 --- a/src/dve/core_engine/backends/implementations/spark/readers/xml.py +++ b/src/dve/core_engine/backends/implementations/spark/readers/xml.py @@ -30,6 +30,7 @@ SparkXMLMode = Literal["PERMISSIVE", "FAILFAST", "DROPMALFORMED"] """The mode to use when parsing XML files with Spark.""" + @spark_filter_null_recs @spark_check_entity_empty @spark_record_index @@ -57,6 +58,7 @@ def read_to_dataframe( ) ) + @spark_filter_null_recs @spark_check_entity_empty @spark_record_index diff --git a/src/dve/core_engine/backends/implementations/spark/spark_helpers.py b/src/dve/core_engine/backends/implementations/spark/spark_helpers.py index f6083fa7..016829d1 100644 --- a/src/dve/core_engine/backends/implementations/spark/spark_helpers.py +++ b/src/dve/core_engine/backends/implementations/spark/spark_helpers.py @@ -639,9 +639,16 @@ def get_spark_cast_statement_from_annotation( return _cast_as_spark_type(stmt, type_) if parent_element else stmt raise ValueError(f"No equivalent Spark type for {type_annotation!r}") -def _spark_filter_null_records(self, entity: DataFrame) -> DataFrame: - return entity.dropna(how="all", subset=[cl for cl in entity.columns if not cl == RECORD_INDEX_COLUMN_NAME]) + +def _spark_filter_null_records(self, entity: DataFrame) -> DataFrame: # pylint: disable=W0613 + return entity.dropna( + how="all", subset=[cl for cl in entity.columns if cl != RECORD_INDEX_COLUMN_NAME] + ) + def spark_filter_null_recs(cls): + """Class decorator to add spark method for filtering records where values + are all null (aside from record_index). + """ setattr(cls, "filter_null_records", _spark_filter_null_records) return cls diff --git a/src/dve/core_engine/backends/readers/utilities.py b/src/dve/core_engine/backends/readers/utilities.py index 99ca6abc..86347dce 100644 --- a/src/dve/core_engine/backends/readers/utilities.py +++ b/src/dve/core_engine/backends/readers/utilities.py @@ -66,7 +66,7 @@ def raise_message_bearing_error_on_header_differences( reporting_field="csv_header", error_code=field_check_error_code, error_message=field_check_error_message, - ) + ), ) diff --git a/src/dve/core_engine/backends/utilities.py b/src/dve/core_engine/backends/utilities.py index 01393221..0f03170f 100644 --- a/src/dve/core_engine/backends/utilities.py +++ b/src/dve/core_engine/backends/utilities.py @@ -273,9 +273,11 @@ def polars_record_index(cls): setattr(cls, "drop_record_index", _drop_polars_record_index) return cls + def polars_filter_null_records(entity: pl.DataFrame) -> pl.DataFrame: + """Strip records where all values (aside from record index) are null""" return entity.filter( - ~pl.all_horizontal(pl.col([cl for cl in entity.columns - if not cl == RECORD_INDEX_COLUMN_NAME]).is_null() - ) - ) \ No newline at end of file + ~pl.all_horizontal( + pl.col([cl for cl in entity.columns if cl != RECORD_INDEX_COLUMN_NAME]).is_null() + ) + )