Coverage for tests / test_dependency_yield_scope.py: 100%

1import json 4 ctx1abcd

2from typing import Annotated, Any 4 ctx1abcd

3 

4import pytest 4 ctx1abcd

5from fastapi import APIRouter, Depends, FastAPI, HTTPException 4 ctx1abcd

6from fastapi.exceptions import FastAPIError 4 ctx1abcd

7from fastapi.responses import StreamingResponse 4 ctx1abcd

8from fastapi.testclient import TestClient 4 ctx1abcd

9 

10 

11class Session: 4 ctx1abcd

12 def __init__(self) -> None: 4 ctx1abcd

13 self.open = True 20 ctx1shkteouvwilxfpyjmzgq

14 

15 

16def dep_session() -> Any: 4 ctx1abcd

17 s = Session() 20 ctx1shkteouvwilxfpyjmzgq

18 yield s 20 ctx1shkteouvwilxfpyjmzgq

19 s.open = False 20 ctx1shkteouvwilxfpyjmzgq

20 

21 

22def raise_after_yield() -> Any: 4 ctx1abcd

23 yield 14 ctx1BCOKPLDEQMFGRN

24 raise HTTPException(status_code=503, detail="Exception after yield") 14 ctx1BCOKPLDEQMFGRN

25 

26 

27SessionFuncDep = Annotated[Session, Depends(dep_session, scope="function")] 4 ctx1abcd

28SessionRequestDep = Annotated[Session, Depends(dep_session, scope="request")] 4 ctx1abcd

29SessionDefaultDep = Annotated[Session, Depends(dep_session)] 4 ctx1abcd

30 

31 

32class NamedSession: 4 ctx1abcd

33 def __init__(self, name: str = "default") -> None: 4 ctx1abcd

34 self.name = name 11 ctx1hkernilfjmg

35 self.open = True 11 ctx1hkernilfjmg

36 

37 

38def get_named_session(session: SessionRequestDep, session_b: SessionDefaultDep) -> Any: 4 ctx1abcd

39 assert session is session_b 4 ctx1enfg

40 named_session = NamedSession(name="named") 4 ctx1enfg

41 yield named_session, session_b 4 ctx1enfg

42 named_session.open = False 4 ctx1enfg

43 

44 

45NamedSessionsDep = Annotated[tuple[NamedSession, Session], Depends(get_named_session)] 4 ctx1abcd

46 

47 

48def get_named_func_session(session: SessionFuncDep) -> Any: 4 ctx1abcd

49 named_session = NamedSession(name="named") 4 ctx1hAij

50 yield named_session, session 4 ctx1hAij

51 named_session.open = False 4 ctx1hAij

52 

53 

54def get_named_regular_func_session(session: SessionFuncDep) -> Any: 4 ctx1abcd

55 named_session = NamedSession(name="named") 4 ctx1krlm

56 return named_session, session 4 ctx1krlm

57 

58 

59BrokenSessionsDep = Annotated[ 4 ctx1abcd

60 tuple[NamedSession, Session], Depends(get_named_func_session) 

61] 

62NamedSessionsFuncDep = Annotated[ 4 ctx1abcd

63 tuple[NamedSession, Session], Depends(get_named_func_session, scope="function") 

64] 

65 

66RegularSessionsDep = Annotated[ 4 ctx1abcd

67 tuple[NamedSession, Session], Depends(get_named_regular_func_session) 

68] 

69 

70app = FastAPI() 4 ctx1abcd

71router = APIRouter() 4 ctx1abcd

72 

73 

74@router.get("/") 4 ctx1abcd

75def get_index(): 4 ctx1abcd

76 return {"status": "ok"} 8 ctx1OKPLQMRN

77 

78 

79@app.get("/function-scope") 4 ctx1abcd

80def function_scope(session: SessionFuncDep) -> Any: 4 ctx1abcd

81 def iter_data(): 4 ctx1suwy

82 yield json.dumps({"is_open": session.open}) 4 ctx1suwy

83 

84 return StreamingResponse(iter_data()) 4 ctx1suwy

85 

86 

87@app.get("/request-scope") 4 ctx1abcd

88def request_scope(session: SessionRequestDep) -> Any: 4 ctx1abcd

89 def iter_data(): 4 ctx1tvxz

90 yield json.dumps({"is_open": session.open}) 4 ctx1tvxz

91 

92 return StreamingResponse(iter_data()) 4 ctx1tvxz

93 

94 

95@app.get("/two-scopes") 4 ctx1abcd

96def get_stream_session( 4 ctx1abcd

97 function_session: SessionFuncDep, request_session: SessionRequestDep 

98) -> Any: 

99 def iter_data(): 4 ctx1oHpq

100 yield json.dumps( 4 ctx1oHpq

101 {"func_is_open": function_session.open, "req_is_open": request_session.open} 

102 ) 

103 

104 return StreamingResponse(iter_data()) 4 ctx1oHpq

105 

106 

107@app.get("/sub") 4 ctx1abcd

108def get_sub(sessions: NamedSessionsDep) -> Any: 4 ctx1abcd

109 def iter_data(): 4 ctx1enfg

110 yield json.dumps( 4 ctx1enfg

111 {"named_session_open": sessions[0].open, "session_open": sessions[1].open} 

112 ) 

113 

114 return StreamingResponse(iter_data()) 4 ctx1enfg

115 

116 

117@app.get("/named-function-scope") 4 ctx1abcd

118def get_named_function_scope(sessions: NamedSessionsFuncDep) -> Any: 4 ctx1abcd

119 def iter_data(): 4 ctx1hAij

120 yield json.dumps( 4 ctx1hAij

121 {"named_session_open": sessions[0].open, "session_open": sessions[1].open} 

122 ) 

123 

124 return StreamingResponse(iter_data()) 4 ctx1hAij

125 

126 

127@app.get("/regular-function-scope") 4 ctx1abcd

128def get_regular_function_scope(sessions: RegularSessionsDep) -> Any: 4 ctx1abcd

129 def iter_data(): 4 ctx1krlm

130 yield json.dumps( 4 ctx1krlm

131 {"named_session_open": sessions[0].open, "session_open": sessions[1].open} 

132 ) 

133 

134 return StreamingResponse(iter_data()) 4 ctx1krlm

135 

136 

137app.include_router( 4 ctx1abcd

138 prefix="/router-scope-function", 

139 router=router, 

140 dependencies=[Depends(raise_after_yield, scope="function")], 

141) 

142 

143app.include_router( 4 ctx1abcd

144 prefix="/router-scope-request", 

145 router=router, 

146 dependencies=[Depends(raise_after_yield, scope="request")], 

147) 

148 

149client = TestClient(app) 4 ctx1abcd

150 

151 

152def test_function_scope() -> None: 4 ctx1abcd

153 response = client.get("/function-scope") 4 ctx1suwy

154 assert response.status_code == 200 4 ctx1suwy

155 data = response.json() 4 ctx1suwy

156 assert data["is_open"] is False 4 ctx1suwy

157 

158 

159def test_request_scope() -> None: 4 ctx1abcd

160 response = client.get("/request-scope") 4 ctx1tvxz

161 assert response.status_code == 200 4 ctx1tvxz

162 data = response.json() 4 ctx1tvxz

163 assert data["is_open"] is True 4 ctx1tvxz

164 

165 

166def test_two_scopes() -> None: 4 ctx1abcd

167 response = client.get("/two-scopes") 4 ctx1oHpq

168 assert response.status_code == 200 4 ctx1oHpq

169 data = response.json() 4 ctx1oHpq

170 assert data["func_is_open"] is False 4 ctx1oHpq

171 assert data["req_is_open"] is True 4 ctx1oHpq

172 

173 

174def test_sub() -> None: 4 ctx1abcd

175 response = client.get("/sub") 4 ctx1enfg

176 assert response.status_code == 200 4 ctx1enfg

177 data = response.json() 4 ctx1enfg

178 assert data["named_session_open"] is True 4 ctx1enfg

179 assert data["session_open"] is True 4 ctx1enfg

180 

181 

182def test_broken_scope() -> None: 4 ctx1abcd

183 with pytest.raises( 4 ctx1STUV

184 FastAPIError, 

185 match='The dependency "get_named_func_session" has a scope of "request", it cannot depend on dependencies with scope "function"', 

186 ): 

187 

188 @app.get("/broken-scope") 4 ctx1STUV

189 def get_broken(sessions: BrokenSessionsDep) -> Any: # pragma: no cover 4 ctx1STUV

190 pass 

191 

192 

193def test_named_function_scope() -> None: 4 ctx1abcd

194 response = client.get("/named-function-scope") 4 ctx1hAij

195 assert response.status_code == 200 4 ctx1hAij

196 data = response.json() 4 ctx1hAij

197 assert data["named_session_open"] is False 4 ctx1hAij

198 assert data["session_open"] is False 4 ctx1hAij

199 

200 

201def test_regular_function_scope() -> None: 4 ctx1abcd

202 response = client.get("/regular-function-scope") 4 ctx1krlm

203 assert response.status_code == 200 4 ctx1krlm

204 data = response.json() 4 ctx1krlm

205 assert data["named_session_open"] is True 4 ctx1krlm

206 assert data["session_open"] is False 4 ctx1krlm

207 

208 

209def test_router_level_dep_scope_function() -> None: 4 ctx1abcd

210 response = client.get("/router-scope-function/") 4 ctx1OPQR

211 assert response.status_code == 503 4 ctx1OPQR

212 assert response.json() == {"detail": "Exception after yield"} 4 ctx1OPQR

213 

214 

215def test_router_level_dep_scope_request() -> None: 4 ctx1abcd

216 with TestClient(app, raise_server_exceptions=False) as client: 4 ctx1KLMN

217 response = client.get("/router-scope-request/") 4 ctx1KLMN

218 assert response.status_code == 200 4 ctx1KLMN

219 assert response.json() == {"status": "ok"} 4 ctx1KLMN

220 

221 

222def test_app_level_dep_scope_function() -> None: 4 ctx1abcd

223 app = FastAPI(dependencies=[Depends(raise_after_yield, scope="function")]) 4 ctx1BIDF

224 

225 @app.get("/app-scope-function") 4 ctx1BIDF

226 def get_app_scope_function(): 4 ctx1BIDF

227 return {"status": "ok"} 4 ctx1BIDF

228 

229 with TestClient(app) as client: 4 ctx1BIDF

230 response = client.get("/app-scope-function") 4 ctx1BIDF

231 assert response.status_code == 503 4 ctx1BIDF

232 assert response.json() == {"detail": "Exception after yield"} 4 ctx1BIDF

233 

234 

235def test_app_level_dep_scope_request() -> None: 4 ctx1abcd

236 app = FastAPI(dependencies=[Depends(raise_after_yield, scope="request")]) 4 ctx1CJEG

237 

238 @app.get("/app-scope-request") 4 ctx1CJEG

239 def get_app_scope_request(): 4 ctx1CJEG

240 return {"status": "ok"} 4 ctx1CJEG

241 

242 with TestClient(app, raise_server_exceptions=False) as client: 4 ctx1CJEG

243 response = client.get("/app-scope-request") 4 ctx1CJEG

244 assert response.status_code == 200 4 ctx1CJEG

245 assert response.json() == {"status": "ok"} 4 ctx1CJEG