Contracts & Products¶

def diff(
    discovery: DiscoveryResult,
    output_base: Path,
    models_dir: str | None = None,
    sources_dir: str | None = None,
) -> DiffResult:
    """Compare expected generated output with current on-disk state.

    Args:
        discovery: Result from discover().
        output_base: Base path for output (the dbt project root).
        models_dir: Override for models output directory.
        sources_dir: Override for sources output directory.

    Returns:
        DiffResult with per-file comparison.
    """
    gen_result = generate(
        discovery,
        output_base=output_base,
        models_dir=models_dir,
        sources_dir=sources_dir,
        dry_run=True,
    )

    diffs: list[FileDiff] = []

    for f in gen_result.files:
        diffs.append(_compare_file(f))

    return DiffResult(diffs=diffs)

def discover(contracts_dir: str | Path) -> DiscoveryResult:
    """Scan a contracts directory and load all contracts, products, and config.

    Expects the following structure::

        contracts_dir/
        ├── config.yaml
        ├── contracts/*.odcs.yaml
        └── products/*.odps.yaml

    Args:
        contracts_dir: Path to the contracts directory.

    Returns:
        DiscoveryResult with parsed config, contracts, and products.

    Raises:
        DiscoveryError: If the directory does not exist or a file fails to parse.
    """
    base = Path(contracts_dir)
    if not base.is_dir():
        raise DiscoveryError(f"Contracts directory not found: {base}")

    config = _load_config(base)
    contracts = _load_contracts(base / "contracts")
    products = _load_products(base / "products")

    return DiscoveryResult(
        config=config,
        contracts=contracts,
        products=products,
    )

def generate(
    discovery: DiscoveryResult,
    output_base: Path,
    models_dir: str | None = None,
    sources_dir: str | None = None,
    force: bool = False,
    dry_run: bool = False,
) -> GenerateResult:
    """Generate dbt artifacts from discovered contracts and products.

    Args:
        discovery: Result from discover().
        output_base: Base path for output (typically the dbt project root).
        models_dir: Override for models output directory.
        sources_dir: Override for sources output directory.
        force: Overwrite non-managed files.
        dry_run: Preview without writing files.

    Returns:
        GenerateResult with all generated (or previewed) files.
    """
    gen_config = discovery.config.generation
    resolved_models_dir = output_base / (
        models_dir or (gen_config.models_dir if gen_config else "models/generated")
    )
    resolved_sources_dir = output_base / (
        sources_dir or (gen_config.sources_dir if gen_config else "models/staging")
    )
    header = (
        gen_config.header
        if gen_config
        else "-- Generated by dbt-contracts. Do not edit manually."
    )
    gen_sources = gen_config.generate_sources if gen_config else True
    gen_tests = gen_config.generate_tests if gen_config else True

    default_server_type = discovery.config.default_server_type or "snowflake"

    contracts = [dc.contract for dc in discovery.contracts]
    products = [dp.product for dp in discovery.products]
    rendered = render(contracts, products, default_server_type)

    files: list[GeneratedFile] = []

    # Sources
    if gen_sources:
        for source_dict in rendered.sources:
            source_name = source_dict.get("sources", [{}])[0].get("name", "sources")
            yaml_str = to_yaml(source_dict)
            content = _prepend_yaml_header(yaml_str, header)
            path = resolved_sources_dir / f"{source_name}.yml"
            files.append(GeneratedFile(path=path, content=content))

    # Models
    for model_dict in rendered.models:
        for model in model_dict.get("models", []):
            model_name = model.get("name", "model")
            single_model_dict = {"version": 2, "models": [model]}

            if not gen_tests:
                _strip_tests(single_model_dict)

            yaml_str = to_yaml(single_model_dict)
            content = _prepend_yaml_header(yaml_str, header)
            path = resolved_models_dir / f"{model_name}.yml"
            files.append(GeneratedFile(path=path, content=content))

    # Staging SQL
    for model_name, sql in rendered.staging_sql.items():
        content = _prepend_sql_header(sql, header)
        path = resolved_models_dir / f"{model_name}.sql"
        files.append(GeneratedFile(path=path, content=content))

    # Write or preview
    if not dry_run:
        for f in files:
            if f.path.exists() and not _is_managed_file(f.path, header) and not force:
                f.skipped = True
                continue
            f.path.parent.mkdir(parents=True, exist_ok=True)
            f.path.write_text(f.content, encoding="utf-8")

    return GenerateResult(files=files)

def import_dbt(
    schema_paths: list[Path],
    output_dir: Path,
    server_type: str = "snowflake",
    dry_run: bool = False,
) -> ImportResult:
    """Parse dbt schema YAML files and generate ODCS contract stubs.

    Args:
        schema_paths: Paths to dbt schema.yml files.
        output_dir: Directory to write generated contract files.
        server_type: Default server type for contracts.
        dry_run: Preview without writing files.

    Returns:
        ImportResult with generated contract stubs.
    """
    contracts: list[ImportedContract] = []

    for schema_path in schema_paths:
        data = yaml.safe_load(schema_path.read_text(encoding="utf-8"))
        if not data:
            continue

        for source in data.get("sources", []):
            contract = _source_to_contract(source, server_type)
            name = source.get("name", "source")
            path = output_dir / f"{name}.odcs.yaml"
            content = to_yaml(contract)
            contracts.append(ImportedContract(path=path, content=content))

        for model in data.get("models", []):
            contract = _model_to_contract(model, server_type)
            name = model.get("name", "model")
            path = output_dir / f"{name}.odcs.yaml"
            content = to_yaml(contract)
            contracts.append(ImportedContract(path=path, content=content))

    if not dry_run:
        output_dir.mkdir(parents=True, exist_ok=True)
        for c in contracts:
            c.path.parent.mkdir(parents=True, exist_ok=True)
            c.path.write_text(c.content, encoding="utf-8")

    return ImportResult(contracts=contracts)

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)

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

def to_yaml(data: dict) -> str:
    """Serialize a dict to YAML string."""
    return yaml.dump(data, sort_keys=False, allow_unicode=True)

def validate(discovery: DiscoveryResult) -> ValidationResult:
    """Validate all discovered contracts and products.

    Runs:
    1. ODCS schema validation (via datacontract-cli lint)
    2. Cross-reference validation (ODPS contractId -> ODCS)
    3. Status threshold checks (against config.validation.min_status)

    Args:
        discovery: Result from discover().

    Returns:
        ValidationResult with any issues found.
    """
    issues: list[ValidationIssue] = []

    contract_ids = {dc.contract.id for dc in discovery.contracts if dc.contract.id}

    validation_config = discovery.config.validation
    min_status = validation_config.min_status if validation_config else "draft"
    cross_reference = validation_config.cross_reference if validation_config else True

    # 1. Validate each ODCS contract
    for dc in discovery.contracts:
        lint_result = lint(dc.contract)
        if not lint_result.passed:
            for error in lint_result.errors:
                issues.append(
                    ValidationIssue(
                        path=str(dc.path),
                        contract_id=dc.contract.id,
                        message=error,
                    )
                )

        if min_status and dc.contract.status:
            issue = _check_status(
                str(dc.path), dc.contract.id, dc.contract.status, min_status
            )
            if issue:
                issues.append(issue)

    # 2. Cross-reference: ODPS contractId -> ODCS contracts
    if cross_reference:
        for dp in discovery.products:
            for port_type in ("inputPorts", "outputPorts"):
                for port in getattr(dp.product, port_type, None) or []:
                    if port.contractId and port.contractId not in contract_ids:
                        label = port_type.rstrip("s")
                        issues.append(
                            ValidationIssue(
                                path=str(dp.path),
                                contract_id=dp.product.id,
                                message=(
                                    f"{label} references unknown contract:"
                                    f" {port.contractId}"
                                ),
                            )
                        )

    return ValidationResult(issues=issues)