Snapshot API

Snapshot bundle models.

class sqldbagent.snapshot.models.SnapshotRequestModel(**data)[source]

Bases: BaseModel

Snapshot regeneration request.

Variables:

  • datasource_name – Datasource identifier.

  • schema_name – Schema name captured by the snapshot.

  • sample_size – Sample size used for table profiling.

Parameters:

  • data (Any)

  • datasource_name (str)

  • schema_name (str)

  • sample_size (int)

datasource_name: str
schema_name: str
sample_size: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sqldbagent.snapshot.models.SnapshotBundleModel(**data)[source]

Bases: BaseModel

Normalized persisted snapshot bundle.

Variables:

  • snapshot_id – Stable snapshot identifier.

  • format_version – Snapshot format version.

  • created_at – Snapshot creation timestamp.

  • datasource_name – Datasource identifier.

  • schema_metadata – Normalized schema metadata.

  • relationship_edges – Relationship graph edges derived from foreign keys.

  • profiles – Per-table profiles captured with the snapshot.

  • content_hash – Deterministic content hash for deduplication and drift detection.

  • summary – Generated short summary.

  • regenerate – Request payload that can rebuild the snapshot later.

Parameters:
snapshot_id: str
format_version: int
created_at: datetime
datasource_name: str
schema_metadata: SchemaModel
relationship_edges: list[RelationshipEdgeModel]
profiles: list[TableProfileModel]
content_hash: str | None
summary: str | None
regenerate: SnapshotRequestModel
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sqldbagent.snapshot.models.SnapshotInventoryEntryModel(**data)[source]

Bases: BaseModel

Stored snapshot inventory entry.

Variables:

  • datasource_name – Datasource identifier.

  • schema_name – Captured schema name.

  • snapshot_id – Snapshot identifier.

  • created_at – Snapshot creation timestamp.

  • content_hash – Snapshot content hash.

  • path – Relative snapshot path under the snapshot root.

  • summary – Summary context for the stored snapshot.

Parameters:

  • data (Any)

  • datasource_name (str)

  • schema_name (str)

  • snapshot_id (str)

  • created_at (datetime)

  • content_hash (str | None)

  • path (str)

  • summary (str | None)

datasource_name: str
schema_name: str
snapshot_id: str
created_at: datetime
content_hash: str | None
path: str
summary: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sqldbagent.snapshot.models.TableDiffModel(**data)[source]

Bases: BaseModel

Per-table snapshot diff details.

Variables:

  • table_name – Qualified or unqualified table name.

  • added_columns – Columns present only in the right snapshot.

  • removed_columns – Columns present only in the left snapshot.

  • changed_columns – Columns present in both snapshots but with changed metadata.

  • metadata_changed – Whether non-column metadata changed.

  • profile_changed – Whether the normalized table profile changed.

  • summary – Generated short summary.

Parameters:
table_name: str
added_columns: list[str]
removed_columns: list[str]
changed_columns: list[str]
metadata_changed: bool
profile_changed: bool
summary: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sqldbagent.snapshot.models.SnapshotDiffModel(**data)[source]

Bases: BaseModel

High-level diff between two snapshot bundles.

Variables:

  • left_snapshot_id – Baseline snapshot identifier.

  • right_snapshot_id – Comparison snapshot identifier.

  • left_content_hash – Baseline content hash.

  • right_content_hash – Comparison content hash.

  • added_tables – Tables added in the right snapshot.

  • removed_tables – Tables removed from the right snapshot.

  • changed_tables – Tables present in both snapshots with metadata changes.

  • added_views – Views added in the right snapshot.

  • removed_views – Views removed from the right snapshot.

  • added_relationships – Relationships added in the right snapshot.

  • removed_relationships – Relationships removed from the right snapshot.

  • summary – Generated short summary.

Parameters:
left_snapshot_id: str
right_snapshot_id: str
left_content_hash: str | None
right_content_hash: str | None
added_tables: list[str]
removed_tables: list[str]
changed_tables: list[TableDiffModel]
added_views: list[str]
removed_views: list[str]
added_relationships: list[str]
removed_relationships: list[str]
summary: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Snapshot persistence service.

class sqldbagent.snapshot.service.SnapshotService(*, datasource_name, inspector, profiler, artifacts)[source]

Bases: object

Create, persist, reload, and diff normalized snapshot bundles.

Parameters:
__init__(*, datasource_name, inspector, profiler, artifacts)[source]

Initialize the snapshot service.

Parameters:
Return type:

None

create_schema_snapshot(schema_name, *, sample_size=5)[source]

Create a snapshot bundle for one schema.

Parameters:

  • schema_name (str) – Schema name to capture.

  • sample_size (int, default: 5) – Sample rows per table profile.

Returns:

Snapshot bundle.

Return type:

SnapshotBundleModel

save_snapshot(bundle)[source]

Persist a snapshot bundle to disk and update the inventory index.

Parameters:

bundle (SnapshotBundleModel) – Snapshot bundle to persist.

Returns:

Snapshot file path.

Return type:

Path

static load_snapshot(path)[source]

Load a snapshot bundle from disk.

Parameters:

path (str | Path) – Snapshot file path.

Returns:

Loaded snapshot bundle.

Return type:

SnapshotBundleModel

static diff_snapshots(left, right)[source]

Diff two snapshot bundles.

Parameters:
Returns:

Snapshot diff payload.

Return type:

SnapshotDiffModel

static list_saved_snapshots(artifacts, *, datasource_name=None, schema_name=None)[source]

List saved snapshots from the inventory index.

Parameters:

  • artifacts (ArtifactSettings) – Artifact persistence settings.

  • datasource_name (str | None, default: None) – Optional datasource filter.

  • schema_name (str | None, default: None) – Optional schema filter.

Returns:

Matching saved snapshots.

Return type:

list[SnapshotInventoryEntryModel]

static load_latest_snapshot(artifacts, *, datasource_name, schema_name)[source]

Load the newest saved snapshot for a datasource/schema pair.

Parameters:

  • artifacts (ArtifactSettings) – Artifact persistence settings.

  • datasource_name (str) – Datasource identifier.

  • schema_name (str) – Captured schema name.

Returns:

Latest matching snapshot.

Return type:

SnapshotBundleModel

Raises:

FileNotFoundError – If no matching snapshot exists.

property snapshot_dir: Path

Return the configured snapshot directory.

Returns:

Snapshot directory path.

Return type:

Path

property datasource_name: str

Return the datasource name bound to this service.

property artifacts: ArtifactSettings

Return the artifact settings bound to this service.

load_latest_saved_snapshot(schema_name)[source]

Load the newest saved snapshot for the bound datasource and schema.

Parameters:

schema_name (str) – Schema whose latest snapshot should be loaded.

Returns:

Latest matching saved snapshot.

Return type:

SnapshotBundleModel

property inventory_path: Path

Return the snapshot inventory file path.

Returns:

Inventory index path.

Return type:

Path