Skip to content

Generation

Adapter Layer

The adapter module wraps datacontract-cli exporters and adds lineage-aware rendering from ODPS products.

dbt_contracts.core.adapter

Adapter layer wrapping datacontract-cli exporters for dbt artifact generation.

Provides lineage-aware rendering of ODCS contracts into dbt sources, models, and staging SQL, using ODPS products to determine contract classification.

LintResult

Bases: BaseModel

Result of validating an ODCS contract against the JSON schema.

Source code in src/dbt_contracts/core/adapter.py 
25
26
27
28
29
class LintResult(pyd.BaseModel):
    """Result of validating an ODCS contract against the JSON schema."""

    passed: bool
    errors: list[str] = pyd.Field(default_factory=list)

GenerationResult

Bases: BaseModel

Combined dbt artifacts produced by rendering contracts.

Source code in src/dbt_contracts/core/adapter.py 
32
33
34
35
36
37
class GenerationResult(pyd.BaseModel):
    """Combined dbt artifacts produced by rendering contracts."""

    sources: list[dict] = pyd.Field(default_factory=list)
    models: list[dict] = pyd.Field(default_factory=list)
    staging_sql: dict[str, str] = pyd.Field(default_factory=dict)

lint(contract)

Validate an ODCS contract against the JSON schema.

Delegates to datacontract-cli's lint functionality. Serializes the contract to YAML first, as datacontract-cli only runs full JSON schema validation when receiving a YAML string (not an object).

Source code in src/dbt_contracts/core/adapter.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def lint(contract: OpenDataContractStandard) -> LintResult:
    """Validate an ODCS contract against the JSON schema.

    Delegates to datacontract-cli's lint functionality. Serializes the contract
    to YAML first, as datacontract-cli only runs full JSON schema validation
    when receiving a YAML string (not an object).
    """
    yaml_str = contract.to_yaml()
    run = DataContract(data_contract_str=yaml_str).lint()
    errors = [
        check.reason or check.name
        for check in run.checks
        if check.result.value == "failed"
    ]
    return LintResult(passed=len(errors) == 0, errors=errors)

render(contracts, products, default_server_type='snowflake')

Render ODCS contracts into dbt artifacts using ODPS lineage.

Classifies each contract as source, model, or ref based on ODPS port definitions, then calls the appropriate datacontract-cli exporter.

Parameters:

Name Type Description Default
contracts list[OpenDataContractStandard]

ODCS contract objects to render.

 required
products list[OpenDataProductStandard]

ODPS product objects defining lineage between contracts.

 required
default_server_type str

Fallback server type when a contract has no servers.

 'snowflake'

Returns:

Type Description
GenerationResult

GenerationResult with sources, models, and staging SQL dicts.
Source code in src/dbt_contracts/core/adapter.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def render(
    contracts: list[OpenDataContractStandard],
    products: list[OpenDataProductStandard],
    default_server_type: str = "snowflake",
) -> GenerationResult:
    """Render ODCS contracts into dbt artifacts using ODPS lineage.

    Classifies each contract as source, model, or ref based on ODPS port
    definitions, then calls the appropriate datacontract-cli exporter.

    Args:
        contracts: ODCS contract objects to render.
        products: ODPS product objects defining lineage between contracts.
        default_server_type: Fallback server type when a contract has no servers.

    Returns:
        GenerationResult with sources, models, and staging SQL dicts.
    """
    contracts_by_id = {c.id: c for c in contracts if c.id}
    lineage = _build_lineage(contracts_by_id, products)
    upstream_map = _build_upstream_map(products)
    result = GenerationResult()

    for contract_id, role in lineage.items():
        contract = contracts_by_id[contract_id]
        server_name = _get_server_name(contract)
        server_type = _resolve_server_type(contract, default_server_type)

        if role == "source":
            source_dict = _export_sources(contract, server_name, server_type)
            result.sources.append(source_dict)
        else:
            model_dict = _export_model(contract, server_name, server_type)
            result.models.append(model_dict)

            upstream_ids = upstream_map.get(contract_id, [])
            for schema_obj in contract.schema_ or []:
                sql = _build_staging_sql(
                    contract,
                    schema_obj.name,
                    upstream_ids,
                    contracts_by_id,
                    lineage,
                )
                result.staging_sql[schema_obj.name] = sql

    return result