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

1""" 

2Routes for interacting with saved search objects. 

3""" 

4 

5from typing import List 

6from uuid import UUID 

7 

8from fastapi import Body, Depends, HTTPException, Path, Response, status 

9 

10import prefect.server.api.dependencies as dependencies 

11import prefect.server.models as models 

12import prefect.server.schemas as schemas 

13from prefect.server.database import PrefectDBInterface, provide_database_interface 

14from prefect.server.utilities.server import PrefectRouter 

15from prefect.types._datetime import now 

16 

17router: PrefectRouter = PrefectRouter(prefix="/saved_searches", tags=["SavedSearches"]) 

18 

19 

20@router.put("/") 

21async def create_saved_search( 

22 saved_search: schemas.actions.SavedSearchCreate, 

23 response: Response, 

24 db: PrefectDBInterface = Depends(provide_database_interface), 

25) -> schemas.core.SavedSearch: 

26 """Creates a new saved search from the provided schema. 

27 

28 If a saved search with the same name already exists, the saved search's fields are 

29 replaced. 

30 """ 

31 

32 # hydrate the input model into a full model 

33 saved_search = schemas.core.SavedSearch(**saved_search.model_dump()) 

34 

35 right_now = now("UTC") 

36 

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

38 model = await models.saved_searches.create_saved_search( 

39 session=session, saved_search=saved_search 

40 ) 

41 

42 if model.created >= right_now: 

43 response.status_code = status.HTTP_201_CREATED 

44 

45 return model 

46 

47 

48@router.get("/{id:uuid}") 

49async def read_saved_search( 

50 saved_search_id: UUID = Path(..., description="The saved search id", alias="id"), 

51 db: PrefectDBInterface = Depends(provide_database_interface), 

52) -> schemas.core.SavedSearch: 

53 """ 

54 Get a saved search by id. 

55 """ 

56 async with db.session_context() as session: 

57 saved_search = await models.saved_searches.read_saved_search( 

58 session=session, saved_search_id=saved_search_id 

59 ) 

60 if not saved_search: 

61 raise HTTPException( 

62 status_code=status.HTTP_404_NOT_FOUND, detail="Saved search not found" 

63 ) 

64 return saved_search 

65 

66 

67@router.post("/filter") 

68async def read_saved_searches( 

69 limit: int = dependencies.LimitBody(), 

70 offset: int = Body(0, ge=0), 

71 db: PrefectDBInterface = Depends(provide_database_interface), 

72) -> List[schemas.core.SavedSearch]: 

73 """ 

74 Query for saved searches. 

75 """ 

76 async with db.session_context() as session: 

77 return await models.saved_searches.read_saved_searches( 

78 session=session, 

79 offset=offset, 

80 limit=limit, 

81 ) 

82 

83 

84@router.delete("/{id:uuid}", status_code=status.HTTP_204_NO_CONTENT) 

85async def delete_saved_search( 

86 saved_search_id: UUID = Path(..., description="The saved search id", alias="id"), 

87 db: PrefectDBInterface = Depends(provide_database_interface), 

88) -> None: 

89 """ 

90 Delete a saved search by id. 

91 """ 

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

93 result = await models.saved_searches.delete_saved_search( 

94 session=session, saved_search_id=saved_search_id 

95 ) 

96 if not result: 

97 raise HTTPException( 

98 status_code=status.HTTP_404_NOT_FOUND, detail="Saved search not found" 

99 )