Skip to content

Commit 7797ba9

Browse files
committed
Merge branch 'mcp-data-catalog-1' into 'enterprise'
feat(mcp): add data catalog convenience tools See merge request dkinternal/testgen/dataops-testgen!544
2 parents 72dbb12 + fad9a27 commit 7797ba9

10 files changed

Lines changed: 594 additions & 59 deletions

File tree

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
"""Shared data catalog convenience service.
2+
3+
Generates CREATE TABLE scripts from profiled column metadata and fetches
4+
sample rows from source tables. Used by both the Streamlit UI and MCP tools.
5+
"""
6+
import logging
7+
from dataclasses import dataclass
8+
from typing import Literal
9+
from uuid import UUID
10+
11+
import pandas as pd
12+
13+
from testgen.common.database.database_service import get_flavor_service
14+
from testgen.common.database.flavor.flavor_service import FlavorService
15+
from testgen.common.models.connection import Connection
16+
from testgen.common.models.data_column import CreateScriptColumn, DataColumnChars
17+
from testgen.common.pii_masking import get_pii_columns, mask_source_data_pii
18+
from testgen.ui.services.database_service import fetch_from_target_db
19+
from testgen.utils import to_dataframe
20+
21+
LOG = logging.getLogger("testgen")
22+
23+
24+
@dataclass
25+
class TableSampleResult:
26+
status: Literal["OK", "ND", "ERR"]
27+
message: str | None = None
28+
df: pd.DataFrame | None = None
29+
pii_redacted: bool = False
30+
31+
32+
def render_create_table_script(
33+
schema_name: str,
34+
table_name: str,
35+
columns: list[CreateScriptColumn],
36+
flavor_service: FlavorService,
37+
*,
38+
annotate_changes: bool = False,
39+
) -> str:
40+
"""Render CREATE TABLE DDL from profiled columns, quoting identifiers for the flavor.
41+
42+
Column types use the profiling-derived suggestion, falling back to the original
43+
database type. ``annotate_changes`` appends ``-- WAS <type>`` comments where the
44+
suggestion differs from the original type.
45+
"""
46+
quote = flavor_service.quote_character
47+
table_ref = flavor_service.get_table_ref(schema_name, table_name)
48+
quoted_names = [f"{quote}{col.column_name}{quote}" for col in columns]
49+
50+
name_width = max(len(name) for name in quoted_names)
51+
type_width = max(len(col.datatype_suggestion or col.db_data_type or "") for col in columns)
52+
53+
col_defs = []
54+
for index, (col, name) in enumerate(zip(columns, quoted_names, strict=True)):
55+
col_type = col.datatype_suggestion or col.db_data_type or ""
56+
separator = "" if index == len(columns) - 1 else ","
57+
line = f"{name:<{name_width}} {col_type:<{type_width}}{separator}"
58+
if (
59+
annotate_changes
60+
and col.db_data_type
61+
and col.datatype_suggestion
62+
and col.db_data_type.lower() != col.datatype_suggestion.lower()
63+
):
64+
line = f"{line} -- WAS {col.db_data_type}"
65+
col_defs.append(line.rstrip())
66+
67+
body = "\n ".join(col_defs)
68+
return f"CREATE TABLE {table_ref} (\n {body}\n);"
69+
70+
71+
def build_create_table_script(
72+
table_group_id: UUID, table_name: str, *, annotate_changes: bool = False,
73+
) -> str | None:
74+
"""Build a CREATE TABLE script for a profiled table, or ``None`` if it is not in the catalog."""
75+
schema_name, columns = DataColumnChars.list_for_create_script(table_group_id, table_name)
76+
if not columns or schema_name is None:
77+
return None
78+
connection = Connection.get_by_table_group(table_group_id)
79+
if connection is None:
80+
return None
81+
flavor_service = get_flavor_service(connection.sql_flavor)
82+
return render_create_table_script(
83+
schema_name, table_name, columns, flavor_service, annotate_changes=annotate_changes,
84+
)
85+
86+
87+
def fetch_table_sample(
88+
connection: Connection,
89+
table_group_id: UUID,
90+
schema_name: str,
91+
table_name: str,
92+
*,
93+
limit: int,
94+
mask_pii: bool,
95+
column_name: str | None = None,
96+
) -> TableSampleResult:
97+
"""Fetch distinct sample rows from a source table, masking PII columns when requested."""
98+
flavor_service = get_flavor_service(connection.sql_flavor)
99+
prefix, suffix = flavor_service.row_limit_clauses(limit)
100+
quote = flavor_service.quote_character
101+
table_ref = flavor_service.get_table_ref(schema_name, table_name)
102+
columns_expr = f"{quote}{column_name}{quote}" if column_name else "*"
103+
query = f"SELECT DISTINCT {prefix} {columns_expr} FROM {table_ref} {suffix}".strip()
104+
105+
try:
106+
results = fetch_from_target_db(connection, query)
107+
except Exception:
108+
LOG.exception("Table sample fetch encountered an error.")
109+
return TableSampleResult("ERR", message="The sample data could not be loaded.")
110+
111+
if not results:
112+
return TableSampleResult("ND")
113+
114+
df = to_dataframe(results)
115+
pii_redacted = False
116+
if mask_pii:
117+
pii_columns = get_pii_columns(str(table_group_id), schema_name, table_name)
118+
pii_redacted = mask_source_data_pii(df, pii_columns)
119+
return TableSampleResult("OK", df=df, pii_redacted=pii_redacted)

testgen/common/models/data_column.py

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -236,6 +236,13 @@ class ColumnSearchHit(EntityMinimal):
236236
column_name: str
237237

238238

239+
@dataclass
240+
class CreateScriptColumn:
241+
column_name: str
242+
db_data_type: str | None
243+
datatype_suggestion: str | None
244+
245+
239246
class DataColumnChars(Entity):
240247
__tablename__ = "data_column_chars"
241248

@@ -268,6 +275,52 @@ class DataColumnChars(Entity):
268275
# warnings_7_days_prior, warnings_30_days_prior, valid_profile_issue_ct,
269276
# valid_test_issue_ct
270277

278+
@classmethod
279+
def list_for_create_script(
280+
cls, table_groups_id: UUID, table_name: str,
281+
) -> tuple[str | None, list[CreateScriptColumn]]:
282+
"""Return ``(schema_name, columns)`` for a table's CREATE TABLE script.
283+
284+
Columns are ordered by ordinal position and carry the profiling-derived type
285+
suggestion from their latest complete profiling run. Returns ``(None, [])`` when
286+
the table is not in the table group's profiled catalog.
287+
"""
288+
query = (
289+
select(
290+
cls.schema_name,
291+
cls.column_name,
292+
cls.db_data_type,
293+
ProfileResult.datatype_suggestion,
294+
)
295+
.outerjoin(
296+
ProfileResult,
297+
and_(
298+
ProfileResult.profile_run_id == cls.last_complete_profile_run_id,
299+
ProfileResult.schema_name == cls.schema_name,
300+
ProfileResult.table_name == cls.table_name,
301+
ProfileResult.column_name == cls.column_name,
302+
),
303+
)
304+
.where(
305+
cls.table_groups_id == table_groups_id,
306+
cls.table_name == table_name,
307+
cls.drop_date.is_(None),
308+
)
309+
.order_by(asc(cls.ordinal_position), asc(cls.column_name))
310+
)
311+
rows = get_current_session().execute(query).mappings().all()
312+
if not rows:
313+
return None, []
314+
columns = [
315+
CreateScriptColumn(
316+
column_name=row["column_name"],
317+
db_data_type=row["db_data_type"],
318+
datatype_suggestion=row["datatype_suggestion"],
319+
)
320+
for row in rows
321+
]
322+
return rows[0]["schema_name"], columns
323+
271324
@classmethod
272325
def list_for_table_group(
273326
cls,

testgen/mcp/server.py

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -168,6 +168,7 @@ def build_mcp_server(
168168
get_schema_history,
169169
)
170170
from testgen.mcp.tools.profiling import (
171+
generate_create_table_script,
171172
get_column_frequent_values,
172173
get_column_patterns,
173174
get_column_profile_detail,
@@ -203,7 +204,7 @@ def build_mcp_server(
203204
list_schedules,
204205
update_schedule,
205206
)
206-
from testgen.mcp.tools.source_data import get_source_data, get_source_data_query
207+
from testgen.mcp.tools.source_data import get_source_data, get_source_data_query, get_table_sample
207208
from testgen.mcp.tools.table_groups import (
208209
create_table_group,
209210
preview_table_group,
@@ -281,6 +282,8 @@ def safe_prompt(fn):
281282
safe_tool(list_test_notes)
282283
safe_tool(list_test_types)
283284
safe_tool(get_table)
285+
safe_tool(generate_create_table_script)
286+
safe_tool(get_table_sample)
284287
safe_tool(list_column_profiles)
285288
safe_tool(list_profiling_summaries)
286289
safe_tool(list_profiling_runs)

testgen/mcp/tools/profiling.py

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33

44
from sqlalchemy import func, or_
55

6+
from testgen.common.data_catalog_service import build_create_table_script
67
from testgen.common.models import with_database_session
78
from testgen.common.models.data_column import (
89
SUGGESTED_DATA_TYPE_TO_PREFIX,
@@ -1136,3 +1137,21 @@ def search_columns(
11361137
if footer:
11371138
doc.text(footer)
11381139
return doc.render()
1140+
1141+
1142+
@with_database_session
1143+
@mcp_permission("catalog")
1144+
def generate_create_table_script(table_group_id: str, table_name: str) -> str:
1145+
"""Generate a CREATE TABLE script for a profiled table from its columns and suggested data types.
1146+
1147+
Args:
1148+
table_group_id: UUID of the table group, e.g. from `get_data_inventory`.
1149+
table_name: Table name exactly as stored in TestGen (case-sensitive).
1150+
"""
1151+
tg = resolve_table_group(table_group_id)
1152+
1153+
script = build_create_table_script(tg.id, table_name)
1154+
if script is None:
1155+
raise MCPResourceNotAccessible("Table", table_name)
1156+
1157+
return MdDoc().code_block(script, language="sql").render()

testgen/mcp/tools/source_data.py

Lines changed: 53 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,9 @@
11
from datetime import datetime
22

3+
from testgen.common.data_catalog_service import fetch_table_sample
34
from testgen.common.models import with_database_session
5+
from testgen.common.models.connection import Connection
6+
from testgen.common.models.data_column import DataColumnChars
47
from testgen.common.models.profiling_run import ProfilingRun
58
from testgen.common.models.test_definition import TestDefinition
69
from testgen.common.source_data_service import (
@@ -12,7 +15,13 @@
1215
)
1316
from testgen.mcp.exceptions import MCPResourceNotAccessible, MCPUserError
1417
from testgen.mcp.permissions import get_project_permissions, mcp_permission
15-
from testgen.mcp.tools.common import DocGroup, parse_uuid, resolve_hygiene_issue, validate_limit
18+
from testgen.mcp.tools.common import (
19+
DocGroup,
20+
parse_uuid,
21+
resolve_hygiene_issue,
22+
resolve_table_group,
23+
validate_limit,
24+
)
1625
from testgen.mcp.tools.markdown import MdDoc
1726

1827
_DOC_GROUP = DocGroup.INVESTIGATE
@@ -206,3 +215,46 @@ def get_source_data(
206215
doc.code_block(result.query, language="sql")
207216

208217
return doc.render()
218+
219+
220+
@with_database_session
221+
@mcp_permission("catalog")
222+
def get_table_sample(table_group_id: str, table_name: str, limit: int = 100) -> str:
223+
"""Fetch sample rows from a source table for inspection.
224+
225+
Args:
226+
table_group_id: UUID of the table group, e.g. from `get_data_inventory`.
227+
table_name: Table name exactly as stored in TestGen (case-sensitive).
228+
limit: Maximum rows to return (default 100, max 500).
229+
"""
230+
validate_limit(limit, 500)
231+
tg = resolve_table_group(table_group_id)
232+
233+
schema_name, _ = DataColumnChars.list_for_create_script(tg.id, table_name)
234+
if schema_name is None:
235+
raise MCPResourceNotAccessible("Table", table_name)
236+
237+
connection = Connection.get_by_table_group(tg.id)
238+
if connection is None:
239+
raise MCPResourceNotAccessible("Table", table_name)
240+
241+
mask_pii = not get_project_permissions().has_permission("view_pii", tg.project_code)
242+
result = fetch_table_sample(
243+
connection, tg.id, schema_name, table_name, limit=limit, mask_pii=mask_pii,
244+
)
245+
246+
if result.status == "ERR":
247+
raise MCPUserError(
248+
f"Could not read from the source database for connection `{connection.connection_name}`."
249+
)
250+
251+
doc = MdDoc()
252+
if result.status == "ND":
253+
return doc.text("Table has no rows.").render()
254+
255+
row_count = len(result.df) if result.df is not None else 0
256+
doc.field("Rows returned", row_count)
257+
if result.pii_redacted:
258+
doc.text("_PII columns have been redacted._")
259+
doc.table_from_dataframe(result.df)
260+
return doc.render()

0 commit comments

Comments
 (0)