(
self,
path: str = "/",
*,
method: str = "GET",
headers: dict | Headers | None = None,
query_string: dict | None = None,
scheme: str = "http",
data: AnyStr | None = None,
form: dict | None = None,
json: Any = sentinel,
root_path: str = "",
http_version: str = "1.1",
auth: Authorization | tuple[str, str] | None = None,
)
| 321 | |
| 322 | @asynccontextmanager |
| 323 | async def session_transaction( |
| 324 | self, |
| 325 | path: str = "/", |
| 326 | *, |
| 327 | method: str = "GET", |
| 328 | headers: dict | Headers | None = None, |
| 329 | query_string: dict | None = None, |
| 330 | scheme: str = "http", |
| 331 | data: AnyStr | None = None, |
| 332 | form: dict | None = None, |
| 333 | json: Any = sentinel, |
| 334 | root_path: str = "", |
| 335 | http_version: str = "1.1", |
| 336 | auth: Authorization | tuple[str, str] | None = None, |
| 337 | ) -> AsyncGenerator[SessionMixin, None]: |
| 338 | if self.cookie_jar is None: |
| 339 | raise RuntimeError( |
| 340 | "Session transactions only make sense with cookies enabled." |
| 341 | ) |
| 342 | |
| 343 | if headers is None: |
| 344 | headers = Headers() |
| 345 | elif isinstance(headers, Headers): |
| 346 | headers = headers |
| 347 | elif headers is not None: |
| 348 | headers = Headers(headers) |
| 349 | for cookie in self.cookie_jar: |
| 350 | headers.add("cookie", f"{cookie.name}={cookie.value}") |
| 351 | |
| 352 | original_request_ctx = _cv_request.get(None) |
| 353 | async with self.app.test_request_context( |
| 354 | path, |
| 355 | method=method, |
| 356 | headers=headers, |
| 357 | query_string=query_string, |
| 358 | scheme=scheme, |
| 359 | data=data, |
| 360 | form=form, |
| 361 | json=json, |
| 362 | root_path=root_path, |
| 363 | http_version=http_version, |
| 364 | auth=auth, |
| 365 | ) as ctx: |
| 366 | session_interface = self.app.session_interface |
| 367 | session = await session_interface.open_session(self.app, ctx.request) |
| 368 | if session is None: |
| 369 | raise RuntimeError("Error opening the session. Check the secret_key?") |
| 370 | |
| 371 | token = _cv_request.set(original_request_ctx) |
| 372 | try: |
| 373 | yield session |
| 374 | finally: |
| 375 | _cv_request.reset(token) |
| 376 | |
| 377 | response = self.app.response_class(b"") |
| 378 | if not session_interface.is_null_session(session): |
| 379 | await session_interface.save_session(self.app, session, response) |
| 380 | self.cookie_jar.extract_cookies( |
nothing calls this directly
no test coverage detected