-
-
Notifications
You must be signed in to change notification settings - Fork 203
-
The HTTPX docs suggest using a client instance for production code. To use the same client throughout my Quart project, I create a client instance at the module level as shown below.
# http_client.py import httpx _client: httpx.AsyncClient | None = None def init_client(base_url: str, timeout: float = 20) -> None: """Create the shared httpx.AsyncClient.""" global _client if _client is None: _client = httpx.AsyncClient(base_url=base_url, timeout=timeout) async def close_client() -> None: """Close the shared httpx.AsyncClient.""" global _client if _client is not None: await _client.aclose() _client = None def get_client() -> httpx.AsyncClient: """Return the shared httpx.AsyncClient.""" if _client is None: raise RuntimeError("HTTPX client not initialized.") return _client
In the main module I create the Quart app and routes as shown next. The async HTTP client is initialized by the startup() method and the same client instance is used by all the routes. This example just returns the client response in the route. A more practical example would use the response data for some analysis then return JSON as a result of that analysis.
# main.py from quart import Quart from .http_client import close_client, get_client, init_client app = Quart(__name__) @app.before_serving async def startup(): init_client(base_url="https://httpbin.org", timeout=20) @app.get("/") async def root(): return {"message": "hello there", "routes": ["/sample", "/delay4"]} @app.get("/sample") async def sample(): response = await get_client().get("/json") return response.json() @app.get("/delay4") async def delay4(): response = await get_client().get("/delay/4") return response.json() @app.after_serving async def shutdown(): await close_client() print("\nClosed client and shutdown server")
I would like to get some feedback on whether this is a good approach for using a single HTTP client instance in a Quart project. Is there a better solution that doesn't use global? Should I create a class that acts as a wrapper for the httpx.AsyncClient instead of the module approach I showed above? Is there some other design pattern that is more appropriate for a Quart project?
Beta Was this translation helpful? Give feedback.
All reactions
Replies: 3 comments 11 replies
-
You can generally inject a http client into the current_app like so:
import httpx from quart import Quart, current_app app = Quart(__name__) @app.before_serving async def startup(): app.client = httpx.AsyncClient(base_url="https://httpbin.org", timeout=20) @app.after_serving async def shutdown(): await app.client.aclose() @app.get("/sample") async def sample(): response = await current_app.client.get("/json") return response.json()
I would add that if this is for production use/you'd prefer better performance, it would be worth looking into a better library than httpx, such as pyreqwest which outperforms httpx by a factor of 2-13x. Also httpx has some outstanding issues:
- https://github.com/encode/httpx/issues/3215
- https://github.com/encode/httpx/issues/3734
- https://github.com/encode/httpx/issues/3644
among others.
Beta Was this translation helpful? Give feedback.
All reactions
-
The problem with using app.client is LSP tools like pyright and ty complain about an unresolved attribute because it doesn't know what client is on the app object. The same problem occurs with current_app.client. Basically, the type of the client attribute on the app object cannot be inferred by development tools.
Beta Was this translation helpful? Give feedback.
All reactions
-
Then you write a little getter function that is annotated with the expected type:
def get_httpx() -> AsyncClient: return current_app.client # type: ignore
Also, storing stuff on the app instance isn't correct, I'd assume you'd use the ASGI lifetime[state] section of the scope instead, but don't know enough about ASGI to say exactly how.
Beta Was this translation helpful? Give feedback.
All reactions
-
Then you write a little getter function that is annotated with the expected type:
def get_httpx() -> AsyncClient: return current_app.client # type: ignoreAlso, storing stuff on the app instance isn't correct, I'd assume you'd use the ASGI
lifetime[state]section of the scope instead, but don't know enough about ASGI to say exactly how.
(削除) Storing stuff on the app instance is pretty standard, and per the Quart docs: (削除ここまで) (my bad, this is unintended)
The application context is a reference point for any information that isn’t specifically related to a request. This includes the app itself, the g global object and a url_adapter bound only to the app.
https://quart.palletsprojects.com/en/latest/discussion/contexts/#application-context
For type safety, you could subclass Quart like so:
from quart import Quart from httpx import AsyncClient class App(Quart): client: AsyncClient def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.client = AsyncClient()
and then cast current_app to App when you use it. If you'd rather not cast it everytime, you can also create a stub like this:
from typing import cast from quart import current_app as quart_app current_app = cast(App, quart_app)
and then import your version of current_app for use in routes:
@app.route("/") async def index(): resp = await current_app.client.get("https://some.site/")
Beta Was this translation helpful? Give feedback.
All reactions
-
You've misunderstood the documentation. The current_app proxy is for data controlled by Quart. This is true just as it is in Flask. For additional stuff, you'd use an extension, or store stuff on the context[lifecycle][state].
Not saying what you've done won't work, just that it's not intended.
Beta Was this translation helpful? Give feedback.
All reactions
-
👍 2
-
Huh I guess I have misunderstood; I'm fairly certain I've seen this style of context injection used in other projects though (at least with Flask), even if it is unintended. It does feel idiomatic/ergonomic though, and doesn't require much boilerplate.
Beta Was this translation helpful? Give feedback.
All reactions
-
Here is an example where I define a class that acts as a wrapper for the HTTPX async client. The client attribute holds the httpx.AsyncClient that is used elsewhere in the project.
# client.py import httpx class HTTPClient: """Async HTTP client.""" client: httpx.AsyncClient | None = None def start(self, base_url, timeout: float = 20): self.client = httpx.AsyncClient(base_url=base_url, timeout=timeout) async def stop(self): assert self.client is not None await self.client.aclose() def __call__(self) -> httpx.AsyncClient: print("called client") assert self.client is not None return self.client
This HTTPClient class is used in the main module for the routes.
# main.py from quart import Quart from .client import HTTPClient app = Quart(__name__) http_client = HTTPClient() @app.before_serving async def startup(): http_client.start(base_url="https://httpbin.org", timeout=20) @app.after_serving async def shutdown(): await http_client.stop() print("\nClosed HTTP client and shutdown server") @app.get("/") async def root(): return "hello there" @app.get("/sample") async def sample(): client = http_client() response = await client.get("/json") return response.json()
This provides good type checking support and avoids attaching things to the app object itself.
@davidism and @servusdei2018 What do you think about this approach?
Beta Was this translation helpful? Give feedback.
All reactions
-
It's equivalent to just having a global client.
But in this example I'm not using the global keyword like in the original example. This approach seems like it would be easier to test and debug.
Beta Was this translation helpful? Give feedback.
All reactions
-
You don't need the global keyword to read global variables in non-global scopes, only to write to them. The whole init thing you're doing in the original is unnecessary as well, just create the client globally, and close it at the end. Or look into how Flask and Quart extensions are written to manage resources.
Beta Was this translation helpful? Give feedback.
All reactions
-
You don't need the global keyword to read global variables in non-global scopes, only to write to them. The whole init thing you're doing in the original is unnecessary as well, just create the client globally, and close it at the end.
Can you provide some example code that demonstrates this?
Beta Was this translation helpful? Give feedback.
All reactions
-
Literally exactly what you wrote originally, but without the init global function.
client = AsyncClient() @app.after_serving async def close_client(): await client.close()
Beta Was this translation helpful? Give feedback.
All reactions
-
This is what I would like to avoid. If this code is in a package then the client gets initialized by just importing the package.
Beta Was this translation helpful? Give feedback.
All reactions
-
Here is another example which has the following file structure:
.
├── pyproject.toml
├── README.md
├── src
│ └── quart_httpx2
│ ├── __init__.py
│ └── main.py
├── tests
│ └── test_main.py
└── uv.lock
Below is the main module content. I didn't think I could create the client at the module level because pytest complained about the client being closed before running the tests. But I got around that with some fixtures and asyncio support for pytest.
# main.py import httpx from quart import Quart app = Quart(__name__) client = httpx.AsyncClient(base_url="https://jsonplaceholder.typicode.com") @app.after_serving async def shutdown(): """Clean up resources after serving.""" await client.aclose() print("\nClosed client and shutdown server") @app.get("/") async def root(): """Return server status.""" return {"status": "ok", "message": "Server is running"} @app.get("/users") async def get_users(): """Fetch and return user data from JSONPlaceholder API.""" response = await client.get("https://jsonplaceholder.typicode.com/users") return response.json() @app.get("/posts") async def get_posts(): """Fetch and return posts data from JSONPlaceholder API.""" response = await client.get("https://jsonplaceholder.typicode.com/posts") return response.json()
And here are the tests:
# test_main.py import pytest import pytest_asyncio import respx from httpx import Response from quart_httpx2.main import app MOCK_USERS = [ {"id": 1, "name": "John Doe", "email": "john@example.com"}, {"id": 2, "name": "Jane Doe", "email": "jane@example.com"}, ] MOCK_POSTS = [ {"id": 1, "title": "First Post", "body": "Hello world", "userId": 1}, {"id": 2, "title": "Second Post", "body": "Another post", "userId": 1}, ] @pytest_asyncio.fixture(scope="module") async def client(): test_client = app.test_client() yield test_client @pytest.mark.asyncio async def test_root(client): response = await client.get("/") assert response.status_code == 200 data = await response.get_json() assert data == {"status": "ok", "message": "Server is running"} @pytest.mark.asyncio @respx.mock async def test_get_users(client): respx.get("https://jsonplaceholder.typicode.com/users").mock( return_value=Response(200, json=MOCK_USERS) ) response = await client.get("/users") assert response.status_code == 200 data = await response.get_json() assert isinstance(data, list) assert len(data) == 2 assert data[0]["id"] == 1 assert data[0]["name"] == "John Doe" assert data[0]["email"] == "john@example.com" @pytest.mark.asyncio @respx.mock async def test_get_posts(client): respx.get("https://jsonplaceholder.typicode.com/posts").mock( return_value=Response(200, json=MOCK_POSTS) ) response = await client.get("/posts") assert response.status_code == 200 data = await response.get_json() assert isinstance(data, list) assert len(data) == 2 assert data[0]["id"] == 1 assert data[0]["title"] == "First Post" assert data[0]["body"] == "Hello world"
@davidism I think this example aligns with your earlier comments about defining the client globally. I had tried this before but had issues with writing tests but I seem to have fixed those issues. Anyway, what are your thoughts about this approach?
Beta Was this translation helpful? Give feedback.