Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/ui/schemas.py: 76%

1from typing import TYPE_CHECKING, Any 

2 

3from fastapi import Body, Depends, HTTPException 

4 

5from prefect._internal.compatibility.starlette import status 

6from prefect.logging import get_logger 

7from prefect.server.database import PrefectDBInterface, provide_database_interface 

8from prefect.server.schemas.responses import SchemaValuesValidationResponse 

9from prefect.server.utilities.server import APIRouter 

10from prefect.utilities.schema_tools.hydration import HydrationContext, hydrate 

11from prefect.utilities.schema_tools.validation import ( 

12 CircularSchemaRefError, 

13 ValidationError, 

14 build_error_obj, 

15 is_valid_schema, 

16 preprocess_schema, 

17 validate, 

18) 

19 

20if TYPE_CHECKING: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true

21 import logging 

22 

23router: APIRouter = APIRouter(prefix="/ui/schemas", tags=["UI", "Schemas"]) 

24 

25logger: "logging.Logger" = get_logger("server.api.ui.schemas") 

26 

27 

28@router.post("/validate") 

29async def validate_obj( 

30 json_schema: dict[str, Any] = Body( 

31 ..., 

32 embed=True, 

33 alias="schema", 

34 validation_alias="schema", 

35 json_schema_extra={"additionalProperties": True}, 

36 ), 

37 values: dict[str, Any] = Body( 

38 ..., embed=True, json_schema_extra={"additionalProperties": True} 

39 ), 

40 db: PrefectDBInterface = Depends(provide_database_interface), 

41) -> SchemaValuesValidationResponse: 

42 schema = preprocess_schema(json_schema) 

43 

44 try: 

45 is_valid_schema(schema, preprocess=False) 

46 except ValueError as exc: 

47 raise HTTPException( 

48 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc) 

49 ) 

50 

51 async with db.session_context() as session: 

52 ctx = await HydrationContext.build( 

53 session=session, render_jinja=False, render_workspace_variables=True 

54 ) 

55 

56 hydrated_values = hydrate(values, ctx) 

57 try: 

58 errors = validate(hydrated_values, schema, preprocess=False) 

59 except CircularSchemaRefError: 

60 raise HTTPException( 

61 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, 

62 detail="Invalid schema: Unable to validate schema with circular references.", 

63 ) 

64 except ValidationError as exc: 

65 raise HTTPException( 

66 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, 

67 detail=f"Invalid schema: {exc}", 

68 ) 

69 error_obj = build_error_obj(errors) 

70 

71 return error_obj