feat(v3): implement schema discovery endpoint

This commit is contained in:
Scott Idem
2026-01-06 16:03:54 -05:00
parent 45f6303219
commit 459bd89198

View File

@@ -96,6 +96,84 @@ async def health_check(
return mk_resp(data={"status": "V3 API is healthy!"})
@router.get("/{obj_type}/schema", response_model=Resp_Body_Base, tags=['CRUD v3 Schema (Dev)'])
async def get_obj_schema(
response: Response,
obj_type: str = Path(min_length=2, max_length=50),
view: str = Query('default', description="Select alternative view/model (e.g., enriched, detail)"),
variant: str = Query('base', regex='^(base|in|out|default)$', description="Select model variant"),
account: AccountContext = Depends(get_account_context),
):
"""
Returns the schema for a specific Aether object type.
Combines database column metadata with Pydantic model field definitions.
"""
from app.db_sql import db
from sqlalchemy import text
if obj_type not in obj_type_kv_li:
return mk_resp(data=False, status_code=400, response=response, status_message=f"Object type '{obj_type}' not found.")
obj_cfg = obj_type_kv_li[obj_type]
# Select table/view
table_name = obj_cfg.get(f'tbl_{view}', obj_cfg.get('tbl_default', obj_cfg.get('tbl')))
# Select model variant
model_key = f'mdl_{variant}' if variant != 'base' else 'mdl'
model = obj_cfg.get(model_key, obj_cfg.get('mdl_default', obj_cfg.get('mdl')))
if not table_name:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Table configuration for '{obj_type}' is missing.")
schema_info = {
"object_type": obj_type,
"view": view,
"variant": variant,
"database": {
"table_name": table_name,
"columns": []
},
"model": {
"name": model.__name__ if hasattr(model, '__name__') else str(model),
"fields": {}
}
}
# 1. Get Database Metadata
try:
db_result = db.execute(text(f"DESCRIBE `{table_name}`"))
for row in db_result.fetchall():
schema_info["database"]["columns"].append({
"field": row[0],
"type": row[1],
"nullable": row[2] == 'YES',
"key": row[3],
"default": row[4],
"extra": row[5]
})
except Exception as e:
log.warning(f"Failed to describe table '{table_name}': {e}")
schema_info["database"]["error"] = str(e)
# 2. Get Pydantic Model Metadata
if model and hasattr(model, "__fields__"):
for field_name, field in model.__fields__.items():
field_info = {
"alias": field.alias,
"type": str(field.outer_type_),
"required": field.required,
"default": field.default
}
# Include extra metadata from Field() if present
if field.field_info.description:
field_info["description"] = field.field_info.description
schema_info["model"]["fields"][field_name] = field_info
return mk_resp(data=schema_info, response=response)
# --- TOP LEVEL OBJECTS ---
@router.get('/{obj_type_l1}/{obj_id}', response_model=Resp_Body_Base)