Coverage for app / routers / data.py: 89%

1from typing import Annotated 

2 

3import sqlglot 

4import sqlglot.errors 

5from fastapi import APIRouter, Body, HTTPException, status 

6 

7from app.config import settings 

8from app.models import CreateTranslation 

9from app.translator import merge_sql_strings, parse_query_delimiters, restore_casing 

10 

11router = APIRouter(prefix=settings.API_V1_STR, tags=["Data API"]) 

12 

13 

14@router.post("/translate") 

15async def create_translation(data: Annotated[CreateTranslation, Body()]) -> str: 

16 try: 

17 queries = sqlglot.transpile( 

18 sql=data.sql, 

19 read=data.from_dialect, 

20 write=data.to_dialect, 

21 ) 

22 except sqlglot.errors.ParseError as e: 

23 raise HTTPException( 

24 status_code=status.HTTP_400_BAD_REQUEST, 

25 detail=dict(error="ParseError", message=str(e), **e.errors[0]), 

26 ) 

27 except sqlglot.errors.UnsupportedError as e: 

28 raise HTTPException( 

29 status_code=status.HTTP_400_BAD_REQUEST, 

30 detail=dict(error="UnsupportedError", message=str(e)), 

31 ) 

32 translation = merge_sql_strings(queries, parse_query_delimiters(data.sql)) 

33 sql = restore_casing(data.sql, translation) 

34 return sql