Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/variables.py: 86%
71 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 02:04 +0000
1"""
2Routes for interacting with variable objects
3"""
5from typing import List, Optional
6from uuid import UUID
8import sqlalchemy as sa
9from fastapi import Body, Depends, HTTPException, Path, status
10from sqlalchemy.ext.asyncio import AsyncSession
12from prefect.server import models
13from prefect.server.api.dependencies import LimitBody
14from prefect.server.database import (
15 PrefectDBInterface,
16 orm_models,
17 provide_database_interface,
18)
19from prefect.server.schemas import actions, core, filters, sorting
20from prefect.server.utilities.server import PrefectRouter
23async def get_variable_or_404(
24 session: AsyncSession, variable_id: UUID
25) -> orm_models.Variable:
26 """Returns a variable or raises 404 HTTPException if it does not exist"""
28 variable = await models.variables.read_variable(
29 session=session, variable_id=variable_id
30 )
31 if not variable:
32 raise HTTPException(status_code=404, detail="Variable not found.")
34 return variable
37async def get_variable_by_name_or_404(
38 session: AsyncSession, name: str
39) -> orm_models.Variable:
40 """Returns a variable or raises 404 HTTPException if it does not exist"""
42 variable = await models.variables.read_variable_by_name(session=session, name=name)
43 if not variable:
44 raise HTTPException(status_code=404, detail="Variable not found.")
46 return variable
49router: PrefectRouter = PrefectRouter(
50 prefix="/variables",
51 tags=["Variables"],
52)
55@router.post("/", status_code=status.HTTP_201_CREATED)
56async def create_variable(
57 variable: actions.VariableCreate,
58 db: PrefectDBInterface = Depends(provide_database_interface),
59) -> core.Variable:
60 """
61 Create a variable.
63 For more information, see https://docs.prefect.io/v3/concepts/variables.
64 """
65 async with db.session_context(begin_transaction=True) as session:
66 try:
67 model = await models.variables.create_variable(
68 session=session, variable=variable
69 )
70 except sa.exc.IntegrityError:
71 raise HTTPException(
72 status_code=409,
73 detail=f"A variable with the name {variable.name!r} already exists.",
74 )
76 return core.Variable.model_validate(model, from_attributes=True)
79@router.get("/{id:uuid}")
80async def read_variable(
81 variable_id: UUID = Path(..., alias="id"),
82 db: PrefectDBInterface = Depends(provide_database_interface),
83) -> core.Variable:
84 async with db.session_context() as session:
85 model = await get_variable_or_404(session=session, variable_id=variable_id)
87 return core.Variable.model_validate(model, from_attributes=True)
90@router.get("/name/{name:str}")
91async def read_variable_by_name(
92 name: str = Path(...),
93 db: PrefectDBInterface = Depends(provide_database_interface),
94) -> core.Variable:
95 async with db.session_context() as session:
96 model = await get_variable_by_name_or_404(session=session, name=name)
98 return core.Variable.model_validate(model, from_attributes=True)
101@router.post("/filter")
102async def read_variables(
103 limit: int = LimitBody(),
104 offset: int = Body(0, ge=0),
105 variables: Optional[filters.VariableFilter] = None,
106 sort: sorting.VariableSort = Body(sorting.VariableSort.NAME_ASC),
107 db: PrefectDBInterface = Depends(provide_database_interface),
108) -> List[core.Variable]:
109 async with db.session_context() as session:
110 return await models.variables.read_variables(
111 session=session,
112 variable_filter=variables,
113 sort=sort,
114 offset=offset,
115 limit=limit,
116 )
119@router.post("/count")
120async def count_variables(
121 variables: Optional[filters.VariableFilter] = Body(None, embed=True),
122 db: PrefectDBInterface = Depends(provide_database_interface),
123) -> int:
124 async with db.session_context() as session:
125 return await models.variables.count_variables(
126 session=session,
127 variable_filter=variables,
128 )
131@router.patch("/{id:uuid}", status_code=status.HTTP_204_NO_CONTENT)
132async def update_variable(
133 variable: actions.VariableUpdate,
134 variable_id: UUID = Path(..., alias="id"),
135 db: PrefectDBInterface = Depends(provide_database_interface),
136) -> None:
137 async with db.session_context(begin_transaction=True) as session:
138 updated = await models.variables.update_variable(
139 session=session,
140 variable_id=variable_id,
141 variable=variable,
142 )
143 if not updated:
144 raise HTTPException(status_code=404, detail="Variable not found.")
147@router.patch("/name/{name:str}", status_code=status.HTTP_204_NO_CONTENT)
148async def update_variable_by_name(
149 variable: actions.VariableUpdate,
150 name: str = Path(..., alias="name"),
151 db: PrefectDBInterface = Depends(provide_database_interface),
152) -> None:
153 async with db.session_context(begin_transaction=True) as session:
154 updated = await models.variables.update_variable_by_name(
155 session=session,
156 name=name,
157 variable=variable,
158 )
159 if not updated:
160 raise HTTPException(status_code=404, detail="Variable not found.")
163@router.delete("/{id:uuid}", status_code=status.HTTP_204_NO_CONTENT)
164async def delete_variable(
165 variable_id: UUID = Path(..., alias="id"),
166 db: PrefectDBInterface = Depends(provide_database_interface),
167) -> None:
168 async with db.session_context(begin_transaction=True) as session:
169 deleted = await models.variables.delete_variable(
170 session=session, variable_id=variable_id
171 )
172 if not deleted:
173 raise HTTPException(status_code=404, detail="Variable not found.")
176@router.delete("/name/{name:str}", status_code=status.HTTP_204_NO_CONTENT)
177async def delete_variable_by_name(
178 name: str = Path(...),
179 db: PrefectDBInterface = Depends(provide_database_interface),
180) -> None:
181 async with db.session_context(begin_transaction=True) as session:
182 deleted = await models.variables.delete_variable_by_name(
183 session=session, name=name
184 )
185 if not deleted:
186 raise HTTPException(status_code=404, detail="Variable not found.")