Coverage for tests/test_dependency_yield_scope.py: 100%

1import json 7 ctx1abcdefg

2from typing import Any, Tuple 7 ctx1abcdefg

3 

4import pytest 7 ctx1abcdefg

5from fastapi import APIRouter, Depends, FastAPI, HTTPException 7 ctx1abcdefg

6from fastapi.exceptions import FastAPIError 7 ctx1abcdefg

7from fastapi.responses import StreamingResponse 7 ctx1abcdefg

8from fastapi.testclient import TestClient 7 ctx1abcdefg

9from typing_extensions import Annotated 7 ctx1abcdefg

10 

11 

12class Session: 7 ctx1abcdefg

13 def __init__(self) -> None: 7 ctx1abcdefg

14 self.open = True 42 ctx1JovKhCLpwMiDNqxOjEPryQkFRszSlGTtAUmHVuBWnI

15 

16 

17def dep_session() -> Any: 7 ctx1abcdefg

18 s = Session() 42 ctx1JovKhCLpwMiDNqxOjEPryQkFRszSlGTtAUmHVuBWnI

19 yield s 42 ctx1JovKhCLpwMiDNqxOjEPryQkFRszSlGTtAUmHVuBWnI

20 s.open = False 42 ctx1JovKhCLpwMiDNqxOjEPryQkFRszSlGTtAUmHVuBWnI

21 

22 

23def raise_after_yield() -> Any: 7 ctx1abcdefg

24 yield 28 ctx1XY+#Z0,$12-%34.'56/(78:)9!;*

25 raise HTTPException(status_code=503, detail="Exception after yield") 28 ctx1XY+#Z0,$12-%34.'56/(78:)9!;*

26 

27 

28SessionFuncDep = Annotated[Session, Depends(dep_session, scope="function")] 7 ctx1abcdefg

29SessionRequestDep = Annotated[Session, Depends(dep_session, scope="request")] 7 ctx1abcdefg

30SessionDefaultDep = Annotated[Session, Depends(dep_session)] 7 ctx1abcdefg

31 

32 

33class NamedSession: 7 ctx1abcdefg

34 def __init__(self, name: str = "default") -> None: 7 ctx1abcdefg

35 self.name = name 21 ctx1ovhpwiqxjrykszltAmuBn

36 self.open = True 21 ctx1ovhpwiqxjrykszltAmuBn

37 

38 

39def get_named_session(session: SessionRequestDep, session_b: SessionDefaultDep) -> Any: 7 ctx1abcdefg

40 assert session is session_b 7 ctx1hijklmn

41 named_session = NamedSession(name="named") 7 ctx1hijklmn

42 yield named_session, session_b 7 ctx1hijklmn

43 named_session.open = False 7 ctx1hijklmn

44 

45 

46NamedSessionsDep = Annotated[Tuple[NamedSession, Session], Depends(get_named_session)] 7 ctx1abcdefg

47 

48 

49def get_named_func_session(session: SessionFuncDep) -> Any: 7 ctx1abcdefg

50 named_session = NamedSession(name="named") 7 ctx1opqrstu

51 yield named_session, session 7 ctx1opqrstu

52 named_session.open = False 7 ctx1opqrstu

53 

54 

55def get_named_regular_func_session(session: SessionFuncDep) -> Any: 7 ctx1abcdefg

56 named_session = NamedSession(name="named") 7 ctx1vwxyzAB

57 return named_session, session 7 ctx1vwxyzAB

58 

59 

60BrokenSessionsDep = Annotated[ 7 ctx1abcdefg

61 Tuple[NamedSession, Session], Depends(get_named_func_session) 

62] 

63NamedSessionsFuncDep = Annotated[ 7 ctx1abcdefg

64 Tuple[NamedSession, Session], Depends(get_named_func_session, scope="function") 

65] 

66 

67RegularSessionsDep = Annotated[ 7 ctx1abcdefg

68 Tuple[NamedSession, Session], Depends(get_named_regular_func_session) 

69] 

70 

71app = FastAPI() 7 ctx1abcdefg

72router = APIRouter() 7 ctx1abcdefg

73 

74 

75@router.get("/") 7 ctx1abcdefg

76def get_index(): 7 ctx1abcdefg

77 return {"status": "ok"} 14 ctx1+#,$-%.'/(:);*

78 

79 

80@app.get("/function-scope") 7 ctx1abcdefg

81def function_scope(session: SessionFuncDep) -> Any: 7 ctx1abcdefg

82 def iter_data(): 7 ctx1JLNPRTV

83 yield json.dumps({"is_open": session.open}) 7 ctx1JLNPRTV

84 

85 return StreamingResponse(iter_data()) 7 ctx1JLNPRTV

86 

87 

88@app.get("/request-scope") 7 ctx1abcdefg

89def request_scope(session: SessionRequestDep) -> Any: 7 ctx1abcdefg

90 def iter_data(): 7 ctx1KMOQSUW

91 yield json.dumps({"is_open": session.open}) 7 ctx1KMOQSUW

92 

93 return StreamingResponse(iter_data()) 7 ctx1KMOQSUW

94 

95 

96@app.get("/two-scopes") 7 ctx1abcdefg

97def get_stream_session( 7 ctx1abcdefg

98 function_session: SessionFuncDep, request_session: SessionRequestDep 

99) -> Any: 

100 def iter_data(): 7 ctx1CDEFGHI

101 yield json.dumps( 7 ctx1CDEFGHI

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

103 ) 

104 

105 return StreamingResponse(iter_data()) 7 ctx1CDEFGHI

106 

107 

108@app.get("/sub") 7 ctx1abcdefg

109def get_sub(sessions: NamedSessionsDep) -> Any: 7 ctx1abcdefg

110 def iter_data(): 7 ctx1hijklmn

111 yield json.dumps( 7 ctx1hijklmn

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

113 ) 

114 

115 return StreamingResponse(iter_data()) 7 ctx1hijklmn

116 

117 

118@app.get("/named-function-scope") 7 ctx1abcdefg

119def get_named_function_scope(sessions: NamedSessionsFuncDep) -> Any: 7 ctx1abcdefg

120 def iter_data(): 7 ctx1opqrstu

121 yield json.dumps( 7 ctx1opqrstu

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

123 ) 

124 

125 return StreamingResponse(iter_data()) 7 ctx1opqrstu

126 

127 

128@app.get("/regular-function-scope") 7 ctx1abcdefg

129def get_regular_function_scope(sessions: RegularSessionsDep) -> Any: 7 ctx1abcdefg

130 def iter_data(): 7 ctx1vwxyzAB

131 yield json.dumps( 7 ctx1vwxyzAB

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

133 ) 

134 

135 return StreamingResponse(iter_data()) 7 ctx1vwxyzAB

136 

137 

138app.include_router( 7 ctx1abcdefg

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

140 router=router, 

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

142) 

143 

144app.include_router( 7 ctx1abcdefg

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

146 router=router, 

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

148) 

149 

150client = TestClient(app) 7 ctx1abcdefg

151 

152 

153def test_function_scope() -> None: 7 ctx1abcdefg

154 response = client.get("/function-scope") 7 ctx1JLNPRTV

155 assert response.status_code == 200 7 ctx1JLNPRTV

156 data = response.json() 7 ctx1JLNPRTV

157 assert data["is_open"] is False 7 ctx1JLNPRTV

158 

159 

160def test_request_scope() -> None: 7 ctx1abcdefg

161 response = client.get("/request-scope") 7 ctx1KMOQSUW

162 assert response.status_code == 200 7 ctx1KMOQSUW

163 data = response.json() 7 ctx1KMOQSUW

164 assert data["is_open"] is True 7 ctx1KMOQSUW

165 

166 

167def test_two_scopes() -> None: 7 ctx1abcdefg

168 response = client.get("/two-scopes") 7 ctx1CDEFGHI

169 assert response.status_code == 200 7 ctx1CDEFGHI

170 data = response.json() 7 ctx1CDEFGHI

171 assert data["func_is_open"] is False 7 ctx1CDEFGHI

172 assert data["req_is_open"] is True 7 ctx1CDEFGHI

173 

174 

175def test_sub() -> None: 7 ctx1abcdefg

176 response = client.get("/sub") 7 ctx1hijklmn

177 assert response.status_code == 200 7 ctx1hijklmn

178 data = response.json() 7 ctx1hijklmn

179 assert data["named_session_open"] is True 7 ctx1hijklmn

180 assert data["session_open"] is True 7 ctx1hijklmn

181 

182 

183def test_broken_scope() -> None: 7 ctx1abcdefg

184 with pytest.raises( 7 ctx1=?@[]^_

185 FastAPIError, 

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

187 ): 

188 

189 @app.get("/broken-scope") 7 ctx1=?@[]^_

190 def get_broken(sessions: BrokenSessionsDep) -> Any: # pragma: no cover 7 ctx1=?@[]^_

191 pass 

192 

193 

194def test_named_function_scope() -> None: 7 ctx1abcdefg

195 response = client.get("/named-function-scope") 7 ctx1opqrstu

196 assert response.status_code == 200 7 ctx1opqrstu

197 data = response.json() 7 ctx1opqrstu

198 assert data["named_session_open"] is False 7 ctx1opqrstu

199 assert data["session_open"] is False 7 ctx1opqrstu

200 

201 

202def test_regular_function_scope() -> None: 7 ctx1abcdefg

203 response = client.get("/regular-function-scope") 7 ctx1vwxyzAB

204 assert response.status_code == 200 7 ctx1vwxyzAB

205 data = response.json() 7 ctx1vwxyzAB

206 assert data["named_session_open"] is True 7 ctx1vwxyzAB

207 assert data["session_open"] is False 7 ctx1vwxyzAB

208 

209 

210def test_router_level_dep_scope_function() -> None: 7 ctx1abcdefg

211 response = client.get("/router-scope-function/") 7 ctx1+,-./:;

212 assert response.status_code == 503 7 ctx1+,-./:;

213 assert response.json() == {"detail": "Exception after yield"} 7 ctx1+,-./:;

214 

215 

216def test_router_level_dep_scope_request() -> None: 7 ctx1abcdefg

217 with TestClient(app, raise_server_exceptions=False) as client: 7 ctx1#$%'()*

218 response = client.get("/router-scope-request/") 7 ctx1#$%'()*

219 assert response.status_code == 200 7 ctx1#$%'()*

220 assert response.json() == {"status": "ok"} 7 ctx1#$%'()*

221 

222 

223def test_app_level_dep_scope_function() -> None: 7 ctx1abcdefg

224 app = FastAPI(dependencies=[Depends(raise_after_yield, scope="function")]) 7 ctx1XZ13579

225 

226 @app.get("/app-scope-function") 7 ctx1XZ13579

227 def get_app_scope_function(): 7 ctx1XZ13579

228 return {"status": "ok"} 7 ctx1XZ13579

229 

230 with TestClient(app) as client: 7 ctx1XZ13579

231 response = client.get("/app-scope-function") 7 ctx1XZ13579

232 assert response.status_code == 503 7 ctx1XZ13579

233 assert response.json() == {"detail": "Exception after yield"} 7 ctx1XZ13579

234 

235 

236def test_app_level_dep_scope_request() -> None: 7 ctx1abcdefg

237 app = FastAPI(dependencies=[Depends(raise_after_yield, scope="request")]) 7 ctx1Y02468!

238 

239 @app.get("/app-scope-request") 7 ctx1Y02468!

240 def get_app_scope_request(): 7 ctx1Y02468!

241 return {"status": "ok"} 7 ctx1Y02468!

242 

243 with TestClient(app, raise_server_exceptions=False) as client: 7 ctx1Y02468!

244 response = client.get("/app-scope-request") 7 ctx1Y02468!

245 assert response.status_code == 200 7 ctx1Y02468!

246 assert response.json() == {"status": "ok"} 7 ctx1Y02468!