Python API

This reference explains how to use the Soda Core Python API to generate, test, publish, and verify data contracts using local execution (Soda Core) or remote execution (Soda Agent).

Installation

Install the Soda package for your data source:

pip install -i https://pypi.dev.sodadata.io/simple -U soda-postgres

Replace soda-postgres with the appropriate package for your data source. See the Data source reference for Soda Corefor supported packages and configurations.

Configure Logging

To enable logging for contract operations, use:

from soda_core import configure_logging

# Enable or disable verbose logging
configure_logging(verbose=True)

To access logs after a verification run:

from soda_core.contracts import verify_contracts_on_agent

result = verify_contracts_on_agent(
    contract_file_paths="path/to/contract.yaml", 
    soda_cloud_file_path="path/to/soda_cloud.yml",
)

result.get_logs()

Verify a local Contract with Soda Core

Use verify_contracts_locally to run the verification using local configuration and execution:

from soda_core.contracts import verify_contracts_locally

result = verify_contracts_locally(
    data_source_file_path="path/to/data_source.yml",
    contract_file_paths=["contract1.yaml", "contract2.yaml"],
    soda_cloud_file_path="path/to/soda_cloud.yml",
    variables={"START_DATE": "2024-01-01"},
    publish=False,
)

Parameter

Required

Type

Description

data_source_file_path

Yes

str

Path(s) to data source file Data source reference for Soda Core

contract_file_paths

Yes (or is dataset_identifiers required)

str or list[str]

Path(s) to local contract file(s)

dataset_identifiers

Yes (or is contract_file_paths required)

list[str]

Dataset identifiers to fetch contracts from Soda Cloud

variables

dict

Variables to override in contract

publish

bool

Push results to Soda Cloud

soda_cloud_file_path

No (unless publish is True)

str

Path to Soda Cloud config file

Verify Contract with Soda Agent

Use verify_contracts_on_agent to run verification remotely via Soda Agent.

from soda_core.contracts import verify_contracts_on_agent

result = verify_contracts_on_agent(
    contract_file_paths="path/to/contract.yaml",  # or list of paths
    soda_cloud_file_path="path/to/soda_cloud.yml",
    variables={"COUNTRY": "US"},
    publish=True,
)

Parameter

Required

Type

Description

contract_file_paths

Yes (or is dataset_identifiers required)

str or list[str]

Path(s) to local contract file(s)

dataset_identifiers

Yes (or is contract_file_paths required)

list[str]

Dataset identifiers to fetch contracts from Soda Cloud

variables

dict

Variables to override in contract

publish

bool

Push results to Soda Cloud

soda_cloud_file_path

Yes

str

Path to Soda Cloud config file

blocking_timeout_in_minutes

int

How long to wait for the agent to return results (default: 60 minutes)

Result Object

Running a verification returns a ContractVerificationSessionResult that aggregates results for one or more contracts verified in a single session. Use it to read logs and errors across contracts and to compute high-level pass/fail counts. Each contract gets a ContractVerificationResult per-check CheckResult entries and derived counters.

Contract Verification Session Result

Represents the overall verification session (may include multiple contracts).

Attributes

contract_verification_results: list[ContractVerificationResult] : all per-contract results. See Python API

Methods & Properties

get_logs() -> list[str] : concatenated logs from all contracts.
get_logs_str() -> str : newline-joined logs.
get_errors() -> list[str] / get_errors_str() -> str : aggregated error lines from logs.
number_of_checks: int : sum across contracts.
number_of_checks_passed: int : sum across contracts.
number_of_checks_failed: int : sum across contracts.
has_errors: bool : true if any contract has execution errors.
is_failed: bool : true if any contract has failed checks. Ignores execution errors.
is_passed: bool : true if all contracts have no failed checks. Ignores execution errors.
is_ok: bool : true if all contracts are both not failed and have no errors.
assert_ok() -> ContractVerificationSessionResult : raises SodaException if not ok.

ContractVerificationResult

Immutable outcome for a single contract: status, timing, measurements, and per-check results.

Attributes

contract: Contract : See Contract
data_source: DataSource : See DataSource
data_timestamp: Optional[datetime] : data time context (if any).
started_timestamp: datetime / ended_timestamp: datetime : execution window.
status: ContractVerificationStatus : FAILED, PASSED, ERROR, or UNKNOWN. See ContractVerificationResult
measurements: list[Measurement] — See CheckResult
check_results: list[CheckResult] — See Python API
sending_results_to_soda_cloud_failed: bool : whether publishing failed.
log_records: Optional[list[logging.LogRecord]] : detailed logs.

Methods & Properties

get_logs() -> list[str] / get_logs_str() -> str : pull message text from log records.
get_errors() -> list[str] / get_errors_str() -> str : log lines at error level or higher.
has_errors: bool : true if status is ERROR.
is_failed: bool : true if status is FAILED. (Only checks, not execution errors.)
is_passed: bool : true if status is PASSED. (Ignores execution errors.)
is_ok: bool : not failed and not in error.
number_of_checks: int / number_of_checks_passed: int / number_of_checks_failed: int : counts derived from check_results.

CheckResult

Represents the outcome of a single check in the contract.

Attributes & Convenience

check: Check : See Check
outcome: CheckOutcome : PASSED, FAILED, or NOT_EVALUATED. See CheckOutcome
threshold_value: Optional[float|int] : threshold applied.
diagnostic_metric_values: Optional[dict[str, float|int|str]] : diagnostic metrics.
autogenerate_diagnostics_payload: bool : whether to auto-format diagnostics.
is_passed: bool / is_failed: bool / is_not_evaluated: bool : convenience flags.
outcome_emoticon: str : Unicode for quick visual status.
log_table_row() -> dict : compact, user-friendly log row (includes diagnostics).
log_table_row_diagnostics(verbose: bool=True) -> str : formatted diagnostic string; uses _log_console_format(...) for short form.
diagnostics_to_camel_case() -> dict[str, Any] : converts diagnostic metric keys to camelCase (useful for Soda Cloud).

CheckOutcome

PASSED, FAILED, NOT_EVALUATED.

Measurement

Metric container captured during verification.

Attributes

metric_id: str : unique identifier of the metric.
metric_name: Optional[str] : name of the metric.
value: Any : captured value of the metric.

ContractVerificationStatus

Execution status of a contract. Drives is_failed, is_passed, and has_errors.

Values

UNKNOWN : status not determined.
FAILED : contract executed and one or more checks failed.
PASSED : contract executed and all checks passed.
ERROR : execution error occurred.

Check

Represents the executed check definition (identity, type, threshold, etc.).

Attributes

type: str : check type (e.g., missing, row_count).
name: Optional[str] : ame of the check.
qualifier: Optional[str] : string to distinguish multiple checks of same type.
identity: str : unique identifier of the check, used to correlate results in Soda Cloud. It is automatically generated based on the check path in the contract.
column_name: Optional[str] : column the check applies to (if applicable).
threshold: Optional[Threshold] : threshold configuration for this check.
attributes: dict[str, Any] : metadata attributes (for Soda Cloud, routing, etc.).
location: str : where in the contract YAML the check is defined.

Threshold

Configured numeric boundaries for a check outcome.

Attributes

must_be: Optional[float|int] : value must equal the configured number.
must_not_be: Optional[float|int] : value must not equal the configured number.
must_be_greater_than: Optional[float|int] : value must be strictly greater than.
must_be_greater_than_or_equal: Optional[float|int] : value must be greater than or equal.
must_be_less_than: Optional[float|int] : value must be strictly less than.
must_be_less_than_or_equal: Optional[float|int] : value must be less than or equal.
must_be_between: Optional[dict] : value must be within the specified range.
must_be_not_between: Optional[dict] : value must not be within the specified range.

Contract

Describes the verified contract (dataset identity and source info).

Attributes

data_source_name str : name of the data source (as configured in Soda).
dataset_prefix: str : hierarchical prefix (e.g., db, schema).
dataset_name: str : name of the dataset (table/view).
soda_qualified_dataset_name: str : full qualified dataset identifier.
source: YamlFileContentInfo : reference to YAML source definition. See YamlFileContentInfo

YamlFileContentInfo

Describes where the YAML was loaded from.

Attributes

inline_str: Optional[str] : YAML provided inline.
local_path: Optional[str] : local file path to the YAML.

DataSource

Represents the data source used for verification.

Attributes

name: str : logical name of the data source.
type: str : type of the data source (e.g., Postgres, Snowflake, BigQuery).

Error Handling

assert_ok() (on both ContractVerificationSessionResult and ContractVerificationResult) raises SodaException with a human-readable error string if the verification is not OK.
SodaException carries the contract_verification_result for context.

PreviousGenerate API keys NextRest API

Last updated 12 days ago

Was this helpful?