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

1""" 

2Routes for interacting with block objects. 

3""" 

4 

5from typing import List, Optional 

6from uuid import UUID 

7 

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

9 

10from prefect.server import models, schemas 

11from prefect.server.api import dependencies 

12from prefect.server.database import PrefectDBInterface, provide_database_interface 

13from prefect.server.utilities.server import PrefectRouter 

14 

15router: PrefectRouter = PrefectRouter( 

16 prefix="/block_documents", tags=["Block documents"] 

17) 

18 

19 

20@router.post("/", status_code=status.HTTP_201_CREATED) 

21async def create_block_document( 

22 block_document: schemas.actions.BlockDocumentCreate, 

23 db: PrefectDBInterface = Depends(provide_database_interface), 

24) -> schemas.core.BlockDocument: 

25 """ 

26 Create a new block document. 

27 

28 For more information, see https://docs.prefect.io/v3/concepts/blocks. 

29 """ 

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

31 if block_document.name is not None: 

32 exists = ( 

33 await models.block_documents.block_document_with_unique_values_exists( 

34 session=session, 

35 block_type_id=block_document.block_type_id, 

36 name=block_document.name, 

37 ) 

38 ) 

39 if exists: 

40 raise HTTPException( 

41 status.HTTP_409_CONFLICT, 

42 detail="Block already exists", 

43 ) 

44 

45 try: 

46 return await models.block_documents.create_block_document( 

47 session=session, block_document=block_document 

48 ) 

49 except ValueError as exc: 

50 raise HTTPException( 

51 status_code=status.HTTP_400_BAD_REQUEST, 

52 detail=str(exc), 

53 ) 

54 

55 

56@router.post("/filter") 

57async def read_block_documents( 

58 limit: int = dependencies.LimitBody(), 

59 block_documents: Optional[schemas.filters.BlockDocumentFilter] = None, 

60 block_types: Optional[schemas.filters.BlockTypeFilter] = None, 

61 block_schemas: Optional[schemas.filters.BlockSchemaFilter] = None, 

62 include_secrets: bool = Body( 

63 False, description="Whether to include sensitive values in the block document." 

64 ), 

65 sort: Optional[schemas.sorting.BlockDocumentSort] = Body( 

66 schemas.sorting.BlockDocumentSort.NAME_ASC 

67 ), 

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

69 db: PrefectDBInterface = Depends(provide_database_interface), 

70) -> List[schemas.core.BlockDocument]: 

71 """ 

72 Query for block documents. 

73 """ 

74 async with db.session_context() as session: 

75 result = await models.block_documents.read_block_documents( 

76 session=session, 

77 block_document_filter=block_documents, 

78 block_type_filter=block_types, 

79 block_schema_filter=block_schemas, 

80 include_secrets=include_secrets, 

81 sort=sort, 

82 offset=offset, 

83 limit=limit, 

84 ) 

85 

86 return result 

87 

88 

89@router.post("/count") 

90async def count_block_documents( 

91 block_documents: Optional[schemas.filters.BlockDocumentFilter] = None, 

92 block_types: Optional[schemas.filters.BlockTypeFilter] = None, 

93 block_schemas: Optional[schemas.filters.BlockSchemaFilter] = None, 

94 db: PrefectDBInterface = Depends(provide_database_interface), 

95) -> int: 

96 """ 

97 Count block documents. 

98 """ 

99 async with db.session_context() as session: 

100 result = await models.block_documents.count_block_documents( 

101 session=session, 

102 block_document_filter=block_documents, 

103 block_type_filter=block_types, 

104 block_schema_filter=block_schemas, 

105 ) 

106 

107 return result 

108 

109 

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

111async def read_block_document_by_id( 

112 block_document_id: UUID = Path( 

113 ..., description="The block document id", alias="id" 

114 ), 

115 include_secrets: bool = Query( 

116 False, description="Whether to include sensitive values in the block document." 

117 ), 

118 db: PrefectDBInterface = Depends(provide_database_interface), 

119) -> schemas.core.BlockDocument: 

120 async with db.session_context() as session: 

121 block_document = await models.block_documents.read_block_document_by_id( 

122 session=session, 

123 block_document_id=block_document_id, 

124 include_secrets=include_secrets, 

125 ) 

126 if not block_document: 

127 raise HTTPException(status.HTTP_404_NOT_FOUND, "Block document not found") 

128 return block_document 

129 

130 

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

132async def delete_block_document( 

133 block_document_id: UUID = Path( 

134 ..., description="The block document id", alias="id" 

135 ), 

136 db: PrefectDBInterface = Depends(provide_database_interface), 

137) -> None: 

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

139 result = await models.block_documents.delete_block_document( 

140 session=session, block_document_id=block_document_id 

141 ) 

142 if not result: 

143 raise HTTPException( 

144 status.HTTP_404_NOT_FOUND, detail="Block document not found" 

145 ) 

146 

147 

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

149async def update_block_document_data( 

150 block_document: schemas.actions.BlockDocumentUpdate, 

151 block_document_id: UUID = Path( 

152 ..., description="The block document id", alias="id" 

153 ), 

154 db: PrefectDBInterface = Depends(provide_database_interface), 

155) -> None: 

156 try: 

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

158 result = await models.block_documents.update_block_document( 

159 session=session, 

160 block_document_id=block_document_id, 

161 block_document=block_document, 

162 ) 

163 except ValueError as exc: 

164 raise HTTPException( 

165 status_code=status.HTTP_400_BAD_REQUEST, 

166 detail=str(exc), 

167 ) 

168 

169 if not result: 

170 raise HTTPException( 

171 status.HTTP_404_NOT_FOUND, detail="Block document not found" 

172 )