Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/saved_searches.py: 95%
36 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +0000
1"""
2Routes for interacting with saved search objects.
3"""
5from typing import List 1a
6from uuid import UUID 1a
8from fastapi import Body, Depends, HTTPException, Path, Response, status 1a
10import prefect.server.api.dependencies as dependencies 1a
11import prefect.server.models as models 1a
12import prefect.server.schemas as schemas 1a
13from prefect.server.database import PrefectDBInterface, provide_database_interface 1a
14from prefect.server.utilities.server import PrefectRouter 1a
15from prefect.types._datetime import now 1a
17router: PrefectRouter = PrefectRouter(prefix="/saved_searches", tags=["SavedSearches"]) 1a
20@router.put("/") 1a
21async def create_saved_search( 1a
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.
28 If a saved search with the same name already exists, the saved search's fields are
29 replaced.
30 """
32 # hydrate the input model into a full model
33 saved_search = schemas.core.SavedSearch(**saved_search.model_dump()) 1b
35 right_now = now("UTC") 1b
37 async with db.session_context(begin_transaction=True) as session: 1b
38 model = await models.saved_searches.create_saved_search( 1b
39 session=session, saved_search=saved_search
40 )
42 if model.created >= right_now: 1b
43 response.status_code = status.HTTP_201_CREATED 1b
45 return model 1b
48@router.get("/{id:uuid}") 1a
49async def read_saved_search( 1a
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: 1bc
57 saved_search = await models.saved_searches.read_saved_search( 1bc
58 session=session, saved_search_id=saved_search_id
59 )
60 if not saved_search: 60 ↛ 64line 60 didn't jump to line 64 because the condition on line 60 was always true1bc
61 raise HTTPException( 1bc
62 status_code=status.HTTP_404_NOT_FOUND, detail="Saved search not found"
63 )
64 return saved_search
67@router.post("/filter") 1a
68async def read_saved_searches( 1a
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: 1bdec
77 return await models.saved_searches.read_saved_searches( 1bdec
78 session=session,
79 offset=offset,
80 limit=limit,
81 )
84@router.delete("/{id:uuid}", status_code=status.HTTP_204_NO_CONTENT) 1a
85async def delete_saved_search( 1a
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: 1bc
93 result = await models.saved_searches.delete_saved_search( 1bc
94 session=session, saved_search_id=saved_search_id
95 )
96 if not result: 1bc
97 raise HTTPException( 1bc
98 status_code=status.HTTP_404_NOT_FOUND, detail="Saved search not found"
99 )