URL: http://github.com/graphql-python/graphql-server/pull/129.patch
t_uploads_enabled=multipart_uploads_enabled, + ) + self.view.result_override = result_override + + async def _graphql_request( + self, + method: Literal["get", "post"], + query: Optional[str] = None, + operation_name: Optional[str] = None, + variables: Optional[dict[str, object]] = None, + files: Optional[dict[str, BytesIO]] = None, + headers: Optional[dict[str, str]] = None, + extensions: Optional[dict[str, Any]] = None, + **kwargs: Any, + ) -> ClientResponse: + body = self._build_body( + query=query, + operation_name=operation_name, + variables=variables, + files=files, + method=method, + extensions=extensions, + ) + + data: Union[dict[str, object], str, None] = None + + url = "/graphql" + + if body and files: + body.update({name: (file, name) for name, file in files.items()}) + + if method == "get": + body_encoded = urllib.parse.urlencode(body or {}) + url = f"{url}?{body_encoded}" + else: + if body: + data = body if files else json.dumps(body) + kwargs["body"] = data + + headers = self._get_headers(method=method, headers=headers, files=files) + + return await self.request(url, method, headers=headers, **kwargs) + + def _do_request( + self, + url: str, + method: Literal["get", "post", "patch", "put", "delete"], + headers: Optional[dict[str, str]] = None, + **kwargs: Any, + ) -> ClientResponse: + body = kwargs.get("body", None) + req = Request.blank( + url, method=method.upper(), headers=headers or {}, body=body + ) + resp = self.view.dispatch_request(req) + return ClientResponse( + status_code=resp.status_code, data=resp.body, headers=resp.headers + ) + + async def request( + self, + url: str, + method: Literal["head", "get", "post", "patch", "put", "delete"], + headers: Optional[dict[str, str]] = None, + **kwargs: Any, + ) -> ClientResponse: + loop = asyncio.get_running_loop() + ctx = contextvars.copy_context() + func_call = functools.partial( + ctx.run, self._do_request, url=url, method=method, headers=headers, **kwargs + ) + return await loop.run_in_executor(None, func_call) # type: ignore + + async def get( + self, url: str, headers: Optional[dict[str, str]] = None + ) -> ClientResponse: + return await self.request(url, "get", headers=headers) + + async def post( + self, + url: str, + data: Optional[bytes] = None, + json: Optional[JSON] = None, + headers: Optional[dict[str, str]] = None, + ) -> ClientResponse: + body = json if json is not None else data + return await self.request(url, "post", headers=headers, body=body) diff --git a/src/tests/http/conftest.py b/src/tests/http/conftest.py index cd8b8b6..2b7a563 100644 --- a/src/tests/http/conftest.py +++ b/src/tests/http/conftest.py @@ -18,6 +18,7 @@ def _get_http_client_classes() -> Generator[Any, None, None]: ("DjangoHttpClient", "django", [pytest.mark.django]), ("FastAPIHttpClient", "fastapi", [pytest.mark.fastapi]), ("FlaskHttpClient", "flask", [pytest.mark.flask]), + ("WebobHttpClient", "webob", [pytest.mark.webob]), ("QuartHttpClient", "quart", [pytest.mark.quart]), ("SanicHttpClient", "sanic", [pytest.mark.sanic]), ("LitestarHttpClient", "litestar", [pytest.mark.litestar]), From 1f8bb6da75428ad93a4a41e1b730ccf73e425481 Mon Sep 17 00:00:00 2001 From: Syrus AkbaryNote: This service is not intended for secure transactions such as banking, social media, email, or purchasing. Use at your own risk. We assume no liability whatsoever for broken pages.
Alternative Proxies: