44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
from typing import Optional, Dict, Any, List
|
|
from fastapi import APIRouter
|
|
from pydantic import BaseModel, Field
|
|
from app.services.dbt_service import compile_dbt_model
|
|
|
|
router = APIRouter(tags=["dbt"])
|
|
|
|
|
|
class DbtCompileRequest(BaseModel):
|
|
model_name: str = Field(..., description="Name of the dbt model (e.g. silver_cems_emission_enriched)")
|
|
layer: Optional[str] = Field("silver", description="Medallion layer (silver or gold)")
|
|
query: str = Field(..., description="SQL / Jinja query text to compile and validate")
|
|
test_configs: Optional[Dict[str, Any]] = Field(None, description="Shift-left quality tests (unique, not_null, accepted_values, contracts)")
|
|
|
|
|
|
class DbtCompileResponse(BaseModel):
|
|
status: str
|
|
valid: bool
|
|
model_name: str
|
|
layer: str
|
|
compiled_sql: str
|
|
schema_yml: str
|
|
extracted_refs: List[str] = []
|
|
extracted_sources: List[Dict[str, str]] = []
|
|
warnings: List[str] = []
|
|
errors: List[str] = []
|
|
|
|
|
|
@router.post("/dbt/compile", response_model=DbtCompileResponse)
|
|
def compile_dbt_model_endpoint(req: DbtCompileRequest):
|
|
"""
|
|
Pre-flight compile and validation check for dbt models:
|
|
- Validates Jinja expressions & delimiters.
|
|
- Resolves ref() and source() bindings.
|
|
- Emits compiled SQL and matching dbt schema.yml with test assertions.
|
|
"""
|
|
res = compile_dbt_model(
|
|
model_name=req.model_name,
|
|
layer=req.layer or "silver",
|
|
sql_query=req.query,
|
|
test_configs=req.test_configs
|
|
)
|
|
return res
|