Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/admin.py: 88%

1""" 

2Routes for admin-level interactions with the Prefect REST API. 

3""" 

4 

5from fastapi import Depends, HTTPException, status 

6 

7import prefect 

8import prefect.settings 

9from prefect.server import models, schemas 

10from prefect.server.database import PrefectDBInterface, provide_database_interface 

11from prefect.server.exceptions import ObjectNotFoundError 

12from prefect.server.utilities.server import PrefectRouter 

13 

14router: PrefectRouter = PrefectRouter(prefix="/admin", tags=["Admin"]) 

15 

16 

17@router.get("/settings") 

18async def read_settings() -> prefect.settings.Settings: 

19 """ 

20 Get the current Prefect REST API settings. 

21 

22 Secret setting values will be obfuscated. 

23 """ 

24 return prefect.settings.get_current_settings() 

25 

26 

27@router.get("/version") 

28async def read_version() -> str: 

29 """Returns the Prefect version number""" 

30 return prefect.__version__ 

31 

32 

33@router.get("/storage") 

34async def read_server_default_result_storage( 

35 db: PrefectDBInterface = Depends(provide_database_interface), 

36) -> schemas.core.ServerDefaultResultStorage: 

37 """Get the configured server default result storage block.""" 

38 async with db.session_context() as session: 

39 return await models.storage_defaults.read_server_default_result_storage( 

40 session=session 

41 ) 

42 

43 

44@router.put("/storage") 

45async def update_server_default_result_storage( 

46 configuration: schemas.core.ServerDefaultResultStorageUpdate, 

47 db: PrefectDBInterface = Depends(provide_database_interface), 

48) -> schemas.core.ServerDefaultResultStorage: 

49 """Set the server default result storage block.""" 

50 try: 

51 async with db.session_context(begin_transaction=True) as session: 

52 block_document_id = configuration.default_result_storage_block_id 

53 await models.storage_defaults.validate_server_default_result_storage_block( 

54 session=session, 

55 block_document_id=block_document_id, 

56 ) 

57 

58 await models.storage_defaults.write_server_default_result_storage( 

59 session=session, 

60 storage_default=schemas.core.ServerDefaultResultStorage( 

61 default_result_storage_block_id=block_document_id 

62 ), 

63 ) 

64 

65 return await models.storage_defaults.read_server_default_result_storage( 

66 session=session 

67 ) 

68 except ObjectNotFoundError as exc: 

69 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) 

70 except ValueError as exc: 

71 raise HTTPException( 

72 status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, 

73 detail=str(exc), 

74 ) 

75 

76 

77@router.delete("/storage", status_code=status.HTTP_204_NO_CONTENT) 

78async def clear_server_default_result_storage( 

79 db: PrefectDBInterface = Depends(provide_database_interface), 

80) -> None: 

81 """Clear the configured server default result storage block.""" 

82 async with db.session_context(begin_transaction=True) as session: 

83 await models.storage_defaults.clear_server_default_result_storage( 

84 session=session 

85 )