Using an HTTP client in a Quart project

[wigging]

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?

[servusdei2018]

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:

• Improve async performance. encode/httpx#3215
• The create_ssl_context() call can consume a lot of memory encode/httpx#3734
• Fix compatibility with Python 3.14 encode/httpx#3644

among others.

[wigging]

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.

[davidism]

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.

[servusdei2018]

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/")

[davidism]

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.

[servusdei2018]

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.

[wigging]

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?

[wigging]

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.

[davidism]

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.

[wigging]

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?

[davidism]

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

client = AsyncClient()

@app.after_serving
async def close_client():
    await client.close()

[wigging]

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.

[wigging]

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?