Coverage for app / routers / hypermedia.py: 62%

1from typing import Annotated 

2 

3import sqlglot 

4import sqlglot.dialects 

5import sqlglot.dialects.dialect 

6from fastapi import APIRouter, Form, Header, Query, Request 

7from fastapi.responses import HTMLResponse 

8from fastapi.templating import Jinja2Templates 

9 

10from app.config import settings 

11from app.models import CreateTranslation 

12from app.translator import merge_sql_strings, parse_query_delimiters, restore_casing 

13 

14router = APIRouter(tags=["Hypermedia API"]) 

15 

16templates_dir = "/assets/templates" if settings.ENVIRONMENT == "prod" else "templates" 

17 

18templates = Jinja2Templates(directory=templates_dir) 

19 

20SQLGLOT_FORMAT_SETTINGS = { 

21 "pretty": True, 

22 "leading_comma": False, 

23 "identify": False, 

24 "pad": 4, 

25 "indent": 4, 

26 "max_text_width": 80, 

27 "normalize": True, 

28 "normalize_functions": "lower", 

29} 

30 

31 

32@router.get("/", response_class=HTMLResponse) 

33async def home(request: Request) -> HTMLResponse: 

34 # TODO: add query param translation_id 

35 return templates.TemplateResponse( 

36 request=request, 

37 name="home.html", 

38 context={"dialects": sqlglot.dialects.DIALECTS}, 

39 ) 

40 

41 

42@router.get("/dialect-dropdown", response_class=HTMLResponse) 

43async def dialects_dopdown( 

44 request: Request, 

45 dialect: Annotated[str, Query()], 

46 hx_target: Annotated[str, Header()], 

47) -> HTMLResponse: 

48 return templates.TemplateResponse( 

49 request=request, 

50 name="fragments/dialect-dropdown.html", 

51 context={ 

52 "dialects": sqlglot.dialects.DIALECTS, 

53 "selected_dialect": dialect, 

54 "dialect_type": hx_target.split("-")[0], 

55 }, 

56 ) 

57 

58 

59@router.post("/translate", response_class=HTMLResponse) 

60async def create_translation( 

61 request: Request, 

62 data: Annotated[CreateTranslation, Form()], 

63 hx_target: Annotated[str, Header()], 

64) -> HTMLResponse: 

65 if hx_target == "input-textarea-container": 

66 textarea_id = "input" 

67 else: 

68 textarea_id = "output" 

69 

70 try: 

71 queries = sqlglot.transpile( 

72 sql=data.sql, 

73 read=data.from_dialect, 

74 write=data.to_dialect, 

75 **SQLGLOT_FORMAT_SETTINGS, # type: ignore 

76 ) 

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

78 sql = restore_casing(data.sql, translation) 

79 except sqlglot.errors.ParseError as e: 

80 sql = "ERROR: " + str(e).split("\n")[0] 

81 except sqlglot.errors.UnsupportedError as e: 

82 sql = "ERROR: " + str(e).split("\n")[0] 

83 

84 return f'<textarea id="{textarea_id}-textarea">{sql}</textarea>'