Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/admin.py: 39%
33 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1"""
2Routes for admin-level interactions with the Prefect REST API.
3"""
5from fastapi import Body, Depends, Response, status 1a
7import prefect 1a
8import prefect.settings 1a
9from prefect.server.database import PrefectDBInterface, provide_database_interface 1a
10from prefect.server.utilities.server import PrefectRouter 1a
12router: PrefectRouter = PrefectRouter(prefix="/admin", tags=["Admin"]) 1a
15@router.get("/settings") 1a
16async def read_settings() -> prefect.settings.Settings: 1a
17 """
18 Get the current Prefect REST API settings.
20 Secret setting values will be obfuscated.
21 """
22 return prefect.settings.get_current_settings()
25@router.get("/version") 1a
26async def read_version() -> str: 1a
27 """Returns the Prefect version number"""
28 return prefect.__version__
31@router.post("/database/clear", status_code=status.HTTP_204_NO_CONTENT) 1a
32async def clear_database( 1a
33 db: PrefectDBInterface = Depends(provide_database_interface),
34 confirm: bool = Body(
35 False,
36 embed=True,
37 description="Pass confirm=True to confirm you want to modify the database.",
38 ),
39 response: Response = None, # type: ignore
40) -> None:
41 """Clear all database tables without dropping them."""
42 if not confirm:
43 response.status_code = status.HTTP_400_BAD_REQUEST
44 return
45 async with db.session_context(begin_transaction=True) as session:
46 # work pool has a circular dependency on pool queue; delete it first
47 await session.execute(db.WorkPool.__table__.delete())
48 for table in reversed(db.Base.metadata.sorted_tables):
49 await session.execute(table.delete())
52@router.post("/database/drop", status_code=status.HTTP_204_NO_CONTENT) 1a
53async def drop_database( 1a
54 db: PrefectDBInterface = Depends(provide_database_interface),
55 confirm: bool = Body(
56 False,
57 embed=True,
58 description="Pass confirm=True to confirm you want to modify the database.",
59 ),
60 response: Response = None,
61) -> None:
62 """Drop all database objects."""
63 if not confirm:
64 response.status_code = status.HTTP_400_BAD_REQUEST
65 return
67 await db.drop_db()
70@router.post("/database/create", status_code=status.HTTP_204_NO_CONTENT) 1a
71async def create_database( 1a
72 db: PrefectDBInterface = Depends(provide_database_interface),
73 confirm: bool = Body(
74 False,
75 embed=True,
76 description="Pass confirm=True to confirm you want to modify the database.",
77 ),
78 response: Response = None,
79) -> None:
80 """Create all database objects."""
81 if not confirm:
82 response.status_code = status.HTTP_400_BAD_REQUEST
83 return
85 await db.create_db()