Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

Sign up
Appearance settings

Using an HTTP client in a Quart project #455

wigging started this conversation in General
Discussion options

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?

You must be logged in to vote

Replies: 3 comments 11 replies

Comment options

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:

among others.

You must be logged in to vote
5 replies
Comment options

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.

Comment options

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.

Comment options

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.

(削除) 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/")
Comment options

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.

Comment options

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.

Comment options

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?

You must be logged in to vote
6 replies
Comment options

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.

Comment options

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.

Comment options

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?

Comment options

Literally exactly what you wrote originally, but without the init global function.

client = AsyncClient()
@app.after_serving
async def close_client():
 await client.close()
Comment options

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.

Comment options

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?

You must be logged in to vote
0 replies
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet

AltStyle によって変換されたページ (->オリジナル) /