API Reference

This page is generated from the source docstrings with sphinx.ext.autodoc. It covers the modules that make up the engine described in Architecture: the internal data models, the schema analyzer, the generator factory, the post-generation constraint check, and the defensive fallback manager. The REST surface built on top of these modules is documented separately in /docs (Swagger UI) when the FastAPI app is running; see FastAPI.

Core data models

Core data models for V4 adaptive framework.

class test_data_workbench.core.models.EntityType(*values)

Bases: Enum

Common e-commerce entity types for classification.

USER = 'user'

CUSTOMER = 'customer'

PRODUCT = 'product'

ORDER = 'order'

REVIEW = 'review'

CATEGORY = 'category'

INVENTORY = 'inventory'

PAYMENT = 'payment'

SHIPPING = 'shipping'

ANALYTICS = 'analytics'

UNKNOWN = 'unknown'

class test_data_workbench.core.models.ConstraintType(*values)

Bases: Enum

Database constraint types.

PRIMARY_KEY = 'primary_key'

FOREIGN_KEY = 'foreign_key'

UNIQUE = 'unique'

NOT_NULL = 'not_null'

CHECK = 'check'

INDEX = 'index'

class test_data_workbench.core.models.Column(name: str, data_type: str, nullable: bool = True, default: str | None = None, max_length: int | None = None, constraints: List[ConstraintType] = <factory>, sample_values: List[Any] = <factory>)

Bases: object

Database column metadata.

name: str

data_type: str

nullable: bool = True

default: str | None = None

max_length: int | None = None

constraints: List[ConstraintType]

sample_values: List[Any]

property is_primary_key: bool

property is_foreign_key: bool

class test_data_workbench.core.models.Relationship(from_table: str, to_table: str, from_column: str, to_column: str, relationship_type: str, strength: float = 1.0)

Bases: object

Table relationship metadata.

from_table: str

to_table: str

from_column: str

to_column: str

relationship_type: str

strength: float = 1.0

class test_data_workbench.core.models.Table(name: str, columns: ~typing.List[~test_data_workbench.core.models.Column], row_count: int | None = 0, entity_type: ~test_data_workbench.core.models.EntityType = EntityType.UNKNOWN, relationships: ~typing.List[~test_data_workbench.core.models.Relationship] = <factory>)

Bases: object

Database table metadata.

name: str

columns: List[Column]

row_count: int | None = 0

entity_type: EntityType = 'unknown'

relationships: List[Relationship]

property primary_keys: List[Column]

property foreign_keys: List[Column]

class test_data_workbench.core.models.BusinessRule(rule_id: str, description: str, rule_type: str, affected_columns: ~typing.List[str], confidence: float, examples: ~typing.List[str] = <factory>)

Bases: object

Inferred business rule from data analysis.

rule_id: str

description: str

rule_type: str

affected_columns: List[str]

confidence: float

examples: List[str]

classmethod create(description: str, rule_type: str, columns: List[str], confidence: float = 0.8) → BusinessRule

class test_data_workbench.core.models.PIIColumn(table: str, column: str, category: str)

Bases: object

A column flagged as likely containing personally identifiable information, based on a name-based heuristic (no row data needed).

table: str

column: str

category: str

class test_data_workbench.core.models.SchemaInfo(database_name: str, tables: ~typing.List[~test_data_workbench.core.models.Table], relationships: ~typing.List[~test_data_workbench.core.models.Relationship], business_rules: ~typing.List[~test_data_workbench.core.models.BusinessRule], analysis_metadata: ~typing.Dict[str, ~typing.Any] = <factory>, pii_columns: ~typing.List[~test_data_workbench.core.models.PIIColumn] = <factory>)

Bases: object

Complete database schema analysis results.

database_name: str

tables: List[Table]

relationships: List[Relationship]

business_rules: List[BusinessRule]

analysis_metadata: Dict[str, Any]

pii_columns: List[PIIColumn]

property entity_types: Dict[EntityType, List[Table]]

Group tables by detected entity type.

property table_names: Set[str]

get_table(name: str) → Table | None

Get table by name.

class test_data_workbench.core.models.GeneratorConfig(entity_type: EntityType, table_name: str, dependencies: List[str] = <factory>, generation_rules: Dict[str, ~typing.Any]=<factory>, fallback_config: Dict[str, ~typing.Any] | None=None)

Bases: object

Configuration for a specific entity generator.

entity_type: EntityType

table_name: str

dependencies: List[str]

generation_rules: Dict[str, Any]

fallback_config: Dict[str, Any] | None = None

class test_data_workbench.core.models.ScenarioConfig(name: str, description: str, entities: ~typing.Dict[str, int], relationships: ~typing.List[str] = <factory>, constraints: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

Business scenario configuration.

name: str

description: str

entities: Dict[str, int]

relationships: List[str]

constraints: Dict[str, Any]

class test_data_workbench.core.models.AdaptationResult(schema_info: ~test_data_workbench.core.models.SchemaInfo, generated_code: ~typing.Dict[str, str], config_files: ~typing.Dict[str, str], templates: ~typing.Dict[str, str], analysis_time: float, success: bool, errors: ~typing.List[str] = <factory>, constraint_check_summary: str | None = None)

Bases: object

Results from rapid adaptation workflow.

schema_info: SchemaInfo

generated_code: Dict[str, str]

config_files: Dict[str, str]

templates: Dict[str, str]

analysis_time: float

success: bool

errors: List[str]

constraint_check_summary: str | None = None

class test_data_workbench.core.models.ValidationResult(valid: bool, errors: List[str] = <factory>, warnings: List[str] = <factory>, suggestions: List[str] = <factory>)

Bases: object

Results from team contribution validation.

valid: bool

errors: List[str]

warnings: List[str]

suggestions: List[str]

class test_data_workbench.core.models.DemoScenario(name: str, description: str, required_features: List[str], fallback_features: List[str], demo_data: Dict[str, Any], success_criteria: List[str])

Bases: object

Demonstration scenario configuration.

name: str

description: str

required_features: List[str]

fallback_features: List[str]

demo_data: Dict[str, Any]

success_criteria: List[str]

class test_data_workbench.core.models.DemoStatus(ready_scenarios: List[DemoScenario], blocked_scenarios: List[DemoScenario], fallback_plan: DemoScenario | None, confidence_score: float)

Bases: object

Current demonstration readiness status.

ready_scenarios: List[DemoScenario]

blocked_scenarios: List[DemoScenario]

fallback_plan: DemoScenario | None

confidence_score: float

Schema analyzer

Database schema reverse-engineering for unknown production systems.

test_data_workbench.adaptation.schema_analyzer.classify_column_pii(column_name: str) → str | None

Return the PII category for a column name, or None if it doesn’t match any known heuristic. Pure name-based check - no DB access.

When a name matches keywords from more than one category (e.g. “ip_address” and “email_address” both contain the bounded token “address”, which also belongs to the “address” category), the longest - i.e. most specific - matching keyword wins, so those are categorized as ip_address / email respectively rather than the more generic address.

class test_data_workbench.adaptation.schema_analyzer.SchemaAnalyzer

Bases: object

Reverse-engineer any PostgreSQL e-commerce database schema.

async analyze_production_schema(connection_string: str, metadata_only: bool = False) → SchemaInfo

Auto-discover tables, relationships, constraints, data patterns.

When metadata_only=True, only the SQLAlchemy inspector (schema reflection: get_table_names, get_columns, get_pk_constraint, get_foreign_keys) is used to describe structure. No SELECT is issued against any table’s row data: sample values are skipped, row counts are left as None (unknown), and data-derived business rule detection is skipped entirely. PII column flagging still runs, since it is a pure column-name heuristic.

identify_entity_types(table_names: List[str]) → Dict[str, EntityType]

Classify tables as users, products, orders, reviews, etc.

Generator factory

Auto-create appropriate generators from discovered schema.

class test_data_workbench.adaptation.generator_factory.GeneratorSet(generators: Dict[str, str], dependencies: Dict[str, List[str]], configs: Dict[str, GeneratorConfig], fallback_generators: Dict[str, str])

Bases: object

Collection of generated Python classes and their metadata.

generators: Dict[str, str]

dependencies: Dict[str, List[str]]

configs: Dict[str, GeneratorConfig]

fallback_generators: Dict[str, str]

class test_data_workbench.adaptation.generator_factory.RelationshipManager(relationships: List[Relationship], dependency_order: List[str], reference_cache: Dict[str, List[Any]])

Bases: object

Manages foreign keys and constraints across generators.

relationships: List[Relationship]

dependency_order: List[str]

reference_cache: Dict[str, List[Any]]

class test_data_workbench.adaptation.generator_factory.GeneratorFactory

Bases: object

Auto-create appropriate generators from discovered schema.

create_generators_from_schema(schema: SchemaInfo) → GeneratorSet

Generate Python classes for each discovered entity type.

build_relationship_handlers(relationships: List[Relationship]) → RelationshipManager

Create foreign key and constraint management system.

Constraint check

Lightweight post-generation constraint validation.

Runs a small, fixed set of structural checks over an already-generated dataset - primary-key uniqueness, non-null columns populated, and foreign-key values referencing generated parent ids - without re-implementing a general-purpose schema validator. Deliberately scoped tight: these are the checks that are cheap to verify and meaningful regardless of entity type, and they are exactly the ones a hand-rolled generator is most likely to get wrong (duplicate ids, missing required fields, foreign keys pointing at nothing).

class test_data_workbench.adaptation.constraint_check.ConstraintCheckResult(checks_passed: int = 0, checks_failed: int = 0, violations: List[str] = <factory>)

Bases: object

Outcome of running check_constraints() over a generated dataset. Each check is one (table, column) or (table, relationship) assertion - not one check per record - so the counts stay meaningful for a “N checks passed” style summary.

checks_passed: int = 0

checks_failed: int = 0

violations: List[str]

property total_checks: int

property all_passed: bool

summary_line() → str

One-line summary suitable for CLI output.

test_data_workbench.adaptation.constraint_check.check_constraints(schema: SchemaInfo, dataset: Dict[str, List[Dict[str, Any]]]) → ConstraintCheckResult

Validate a generated dataset against the schema’s discoverable constraints.

Parameters:

• schema – The analyzed schema (tables, columns, relationships).

• dataset – table name -> list of generated record dicts, e.g. the output of calling a generated *Generator.generate(n) for each table.

Checks performed:

• Primary-key uniqueness: no duplicate non-null primary key values within a table.

• Not-null: every non-nullable column is populated (not None) in every record.

• Foreign-key range: for every declared relationship, every non-null foreign key value in the child table is found among the referenced column’s generated values in the parent table.

A table present in schema but absent from dataset is simply skipped - there is nothing to check. Likewise, a relationship is skipped when either side of it has no data in dataset, since there is no parent id set to check the foreign key against.

test_data_workbench.adaptation.constraint_check.generate_sample_dataset(schema: SchemaInfo, generated_code: Dict[str, str], sample_size: int = 20) → Dict[str, List[Dict[str, Any]]]

Execute generated generator source, in dependency order, to build a small in-memory sample dataset - purely so check_constraints() has something real to check. Nothing here is written to disk.

Parent tables are generated first so their primary key values can be threaded through as reference_data to child generators that accept it (the order/review-style templates), matching how GeneratorFactory._create_order_generator() and friends expect to be wired up.

Best effort by design: a table whose generator source fails to exec, defines no *Generator class, or raises while generating, is simply skipped rather than aborting the whole check - a single bad generator should not take down deploy’s constraint reporting.

Defensive generator manager

Defensive generator management with automatic fallbacks and error isolation.

class test_data_workbench.core.defensive_manager.GeneratorResult(data: List[Dict[str, Any]], success: bool, generator_used: str, errors: List[str], warnings: List[str])

Bases: object

Result from generator execution with error handling.

data: List[Dict[str, Any]]

success: bool

generator_used: str

errors: List[str]

warnings: List[str]

class test_data_workbench.core.defensive_manager.GeneratorInfo(generator_class: Type, fallback_class: Type | None, simple_fallback: Callable, entity_type: EntityType, dependencies: List[str])

Bases: object

Information about a registered generator.

generator_class: Type

fallback_class: Type | None

simple_fallback: Callable

entity_type: EntityType

dependencies: List[str]

class test_data_workbench.core.defensive_manager.DefensiveGeneratorManager(feature_flags: FeatureFlags | None = None)

Bases: object

Manages generators with automatic fallbacks and error isolation.

generators: Dict[str, GeneratorInfo]

reference_data: Dict[str, List[Any]]

error_counts: Dict[str, int]

performance_stats: Dict[str, List[float]]

register_generator(entity_name: str, generator_class: Type, fallback_class: Type | None = None, entity_type: EntityType = EntityType.UNKNOWN, dependencies: List[str] = None) → None

Register a generator with optional fallback.

generate_safely(entity_type: str, count: int, **kwargs) → GeneratorResult

Generate data with automatic fallback to simple versions on failure.

generate_with_dependencies(entity_type: str, count: int) → GeneratorResult

Generate data respecting dependency order.

batch_generate(entity_counts: Dict[str, int]) → Dict[str, GeneratorResult]

Generate multiple entities in dependency order.

get_health_report() → Dict[str, Any]

Get health report showing error rates and performance.