Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/saved_searches.py: 43%
36 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +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())
35 right_now = now("UTC")
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 )
42 if model.created >= right_now:
43 response.status_code = status.HTTP_201_CREATED
45 return model
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:
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
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:
77 return await models.saved_searches.read_saved_searches(
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:
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 )