Jun 6, 2025 · Jun 6, 2025 · Jun 9, 2025 · Jun 11, 2025 · Jun 12, 2025 · Jun 6, 2025
diff --git a/soda-core/src/soda_core/model/data_source/data_source.py b/soda-core/src/soda_core/model/data_source/data_source.py
 import abc
 from typing import Literal
 from typing import Literal, Optional

 from pydantic import BaseModel, Field
 from soda_core.model.data_source.data_source_connection_properties import (
    DataSourceConnectionProperties,
 )
 from soda_core.model.failed_rows import FailedRowsConfigDatasource


 class DataSourceBase(
    connection_properties: DataSourceConnectionProperties = Field(
        ..., alias="connection", description="Data source connection details"
    )
    failed_rows: Optional[FailedRowsConfigDatasource] = Field(..., description="Configuration for failed rows storage")

    @classmethod
    def get_class_type(cls) -> str:
diff --git a/soda-core/src/soda_core/model/failed_rows.py b/soda-core/src/soda_core/model/failed_rows.py
 import abc
 from enum import Enum
 from typing import Optional

 from pydantic import BaseModel, Field

 DEFAULT_SCHEMA = "failed_rows"


 class FailedRowsStrategy(str, Enum):
    """Enum for failed rows strategy"""

    NONE = "none"
    STORE_DIAGNOSTICS = "store_diagnostics"
    STORE_KEYS = "store_keys"
    STORE_DATA = "store_data"


 class FailedRowsConfigBase(
    BaseModel,
    abc.ABC,
    frozen=True,
    extra="forbid",
    validate_by_name=True,  # Allow to use both field names and aliases when populating from dict
 ):
    enabled: Optional[bool] = Field(
        None,
        description="Enable or disable the storage of failed rows. " "If set to false, failed rows will not be stored.",
    )


 class FailedRowsConfigOrganisation(FailedRowsConfigBase):
    """Top-level configuration for failed rows, used on Soda Cloud."""

    # Overrides the `enabled` field on other levels as Cloud config will return a bool always.
    enabled: bool = Field(
        ...,
        description="Enable or disable the storage of failed rows. " "If set to false, failed rows will not be stored.",
    )

    path_default: Optional[str] = Field(
        "{{ data_source.database }}/{DEFAULT_SCHEMA}",  # TODO: revisit
        description="Path to the warehouse location where failed rows will be stored. ",
    )
    enabled_by_default: bool = Field(
        True,
        description="Enable or disable the storage of failed rows by default. Does not override the `enabled` setting if `enabled` is set to false."
        "If set to false, failed rows will not be stored unless explicitly enabled in the contract or check.",
    )
    strategy_default: FailedRowsStrategy = Field(
        FailedRowsStrategy.STORE_DIAGNOSTICS, description="Default strategy for storing failed rows."
    )


 class FailedRowsConfigDatasource(FailedRowsConfigBase):
    """Top-level configuration for failed rows, on data source level."""

    path: Optional[str] = Field(..., description="Path to the warehouse location where failed rows will be stored.")
    enabled_by_default: Optional[bool] = Field(
        True,
        description="Enable or disable the storage of failed rows by default. Does not override the `enabled` setting if `enabled` is set to false."
        "If set to false, failed rows will not be stored unless explicitly enabled in the contract or check.",
    )
    strategy_default: Optional[FailedRowsStrategy] = Field(
        FailedRowsStrategy.STORE_DIAGNOSTICS, description="Default strategy for storing failed rows."
    )


 class FailedRowsConfigContract(FailedRowsConfigBase):
    """Configuration for failed rows at the contract level."""

    path: Optional[str] = Field(..., description="Path to the warehouse location where failed rows will be stored.")
    strategy: Optional[FailedRowsStrategy] = Field(..., description="Strategy for storing failed rows.")


 class FailedRowsConfigCheck(FailedRowsConfigBase):
    """Configuration for failed rows at the check level."""

    strategy: Optional[FailedRowsStrategy] = Field(..., description="Strategy for storing failed rows.")
Original file line number	Diff line number	Diff line change
		@@ -1,10 +1,11 @@
		import abc
		from typing import Literal
		from typing import Literal, Optional

		from pydantic import BaseModel, Field
		from soda_core.model.data_source.data_source_connection_properties import (
		DataSourceConnectionProperties,
		)
		from soda_core.model.failed_rows import FailedRowsConfigDatasource


		class DataSourceBase(
Expand All		@@ -20,6 +21,7 @@ class DataSourceBase(
		connection_properties: DataSourceConnectionProperties = Field(
		..., alias="connection", description="Data source connection details"
		)
		failed_rows: Optional[FailedRowsConfigDatasource] = Field(..., description="Configuration for failed rows storage")

		@classmethod
		def get_class_type(cls) -> str:
Expand Down
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,79 @@
		import abc
		from enum import Enum
		from typing import Optional

		from pydantic import BaseModel, Field

		DEFAULT_SCHEMA = "failed_rows"


		class FailedRowsStrategy(str, Enum):
		"""Enum for failed rows strategy"""

		NONE = "none"
		STORE_DIAGNOSTICS = "store_diagnostics"
		STORE_KEYS = "store_keys"
		STORE_DATA = "store_data"


		class FailedRowsConfigBase(
		BaseModel,
		abc.ABC,
		frozen=True,
		extra="forbid",
		validate_by_name=True, # Allow to use both field names and aliases when populating from dict
		):
		enabled: Optional[bool] = Field(
		None,
		description="Enable or disable the storage of failed rows. " "If set to false, failed rows will not be stored.",
		)


		class FailedRowsConfigOrganisation(FailedRowsConfigBase):
		"""Top-level configuration for failed rows, used on Soda Cloud."""

		# Overrides the `enabled` field on other levels as Cloud config will return a bool always.
		enabled: bool = Field(
		...,
		description="Enable or disable the storage of failed rows. " "If set to false, failed rows will not be stored.",
		)

		path_default: Optional[str] = Field(
		"{{ data_source.database }}/{DEFAULT_SCHEMA}", # TODO: revisit
		description="Path to the warehouse location where failed rows will be stored. ",
		)
		enabled_by_default: bool = Field(
		True,
		description="Enable or disable the storage of failed rows by default. Does not override the `enabled` setting if `enabled` is set to false."
		"If set to false, failed rows will not be stored unless explicitly enabled in the contract or check.",
		)
		strategy_default: FailedRowsStrategy = Field(
		FailedRowsStrategy.STORE_DIAGNOSTICS, description="Default strategy for storing failed rows."
		)


		class FailedRowsConfigDatasource(FailedRowsConfigBase):
		"""Top-level configuration for failed rows, on data source level."""

		path: Optional[str] = Field(..., description="Path to the warehouse location where failed rows will be stored.")
		enabled_by_default: Optional[bool] = Field(
		True,
		description="Enable or disable the storage of failed rows by default. Does not override the `enabled` setting if `enabled` is set to false."
		"If set to false, failed rows will not be stored unless explicitly enabled in the contract or check.",
		)
		strategy_default: Optional[FailedRowsStrategy] = Field(
		FailedRowsStrategy.STORE_DIAGNOSTICS, description="Default strategy for storing failed rows."
		)


		class FailedRowsConfigContract(FailedRowsConfigBase):
		"""Configuration for failed rows at the contract level."""

		path: Optional[str] = Field(..., description="Path to the warehouse location where failed rows will be stored.")
		strategy: Optional[FailedRowsStrategy] = Field(..., description="Strategy for storing failed rows.")


		class FailedRowsConfigCheck(FailedRowsConfigBase):
		"""Configuration for failed rows at the check level."""

		strategy: Optional[FailedRowsStrategy] = Field(..., description="Strategy for storing failed rows.")