Implementation:Sdv dev SDV ProgrammableConstraint Interface
| Knowledge Sources | |
|---|---|
| Domains | Data_Quality, Extensibility |
| Last Updated | 2026-02-14 00:00 GMT |
Overview
Interface specification for user-defined programmable constraints in SDV synthesis workflows.
Description
The ProgrammableConstraint base class defines the interface that users must implement for custom business rule constraints. Users subclass this and implement the required methods. The framework wraps the user class in a ProgrammableConstraintHarness for integration with the synthesizer pipeline. A SingleTableProgrammableConstraint variant is available for single-table-only constraints.
Usage
Subclass ProgrammableConstraint and implement the required methods: transform, get_updated_metadata, reverse_transform, and is_valid. Optionally implement validate, validate_input_data, and fit.
Code Reference
Source Location
- Repository: SDV
- File: sdv/cag/programmable_constraint.py
- Lines: L9-131
Interface Specification
class ProgrammableConstraint:
"""Base class for custom constraints."""
def validate(self, metadata):
"""Validate metadata compatibility. Optional."""
pass
def validate_input_data(self, data):
"""Validate input data compatibility. Optional."""
pass
def fit(self, data, metadata):
"""Learn from data. Optional."""
pass
def transform(self, data):
"""Transform data to encode constraint.
Args:
data (dict[pd.DataFrame]): Input data.
Returns:
dict[pd.DataFrame]: Transformed data.
"""
raise NotImplementedError()
def get_updated_metadata(self, metadata):
"""Describe metadata changes after transformation.
Args:
metadata (Metadata): Input metadata.
Returns:
Metadata: Updated metadata.
"""
raise NotImplementedError()
def reverse_transform(self, transformed_data):
"""Reverse the transformation.
Args:
transformed_data (dict[pd.DataFrame]): Transformed data.
Returns:
dict[pd.DataFrame]: Original format data.
"""
raise NotImplementedError()
def is_valid(self, synthetic_data):
"""Validate synthetic data.
Args:
synthetic_data (dict[pd.DataFrame]): Data to validate.
Returns:
dict[pd.Series]: Boolean Series per table.
"""
raise NotImplementedError()
Import
from sdv.cag import ProgrammableConstraint, SingleTableProgrammableConstraint
I/O Contract
Inputs (transform)
| Name | Type | Required | Description |
|---|---|---|---|
| data | dict[str, pd.DataFrame] | Yes | Multi-table data dictionary |
Outputs (transform)
| Name | Type | Description |
|---|---|---|
| return value | dict[str, pd.DataFrame] | Transformed data dictionary |
Usage Examples
from sdv.cag import ProgrammableConstraint
from copy import deepcopy
class PositiveProfit(ProgrammableConstraint):
"""Custom constraint ensuring profit = revenue - cost > 0."""
def transform(self, data):
data = deepcopy(data)
table = data['financials']
table['profit_ratio'] = table['revenue'] / (table['cost'] + 1)
table = table.drop(columns=['revenue'])
data['financials'] = table
return data
def get_updated_metadata(self, metadata):
metadata = deepcopy(metadata)
# Update metadata to reflect column changes
return metadata
def reverse_transform(self, data):
data = deepcopy(data)
table = data['financials']
table['revenue'] = table['profit_ratio'] * (table['cost'] + 1)
table = table.drop(columns=['profit_ratio'])
data['financials'] = table
return data
def is_valid(self, data):
table = data['financials']
return {'financials': table['revenue'] > table['cost']}
# Usage
constraint = PositiveProfit()
synthesizer.add_constraints([constraint])