Monday, August 14, 2023

Fetching JSON over streaming HTTP

Recently, as part of my work on Azure OpenAI code samples, I've been experimenting with different ways of streaming data from a server into a website. The most well known technique is web sockets, but there are also other approaches, like server-sent events and readable streams. A readable stream is the simplest of the options, and works well if your website only needs to stream a response from the server (i.e. it doesn't need bi-directional streaming).

HTTP streaming in Python

To stream an HTTP response, your backend needs to set the "Transfer Encoding" to "chunked". Most web frameworks provide documentation about streaming responses, such as Flask: Streaming and Quart: Streaming responses. In both Flask and Quart, the response must be a Python generator, so that the server can continually get the next data from the generator until it's exhausted.

This example from the Flask doc streams data from a CSV:

def generate_large_csv():
    def generate():
        for row in iter_all_rows():
            yield f"{','.join(row)}\n"
    return generate(), {"Content-Type": "text/csv"}

This example from the Quart docs is an infinite stream of timestamps:

async def stream_time():
    async def async_generator():
        time = datetime.isoformat()
        yield time.encode()
    return async_generator(), 200

Consuming streams in JavaScript

The standard way to consume HTTP requests in JavaScript is the fetch() function, and fortunately, that function can also be used to consume HTTP streams. When the browser sees that the data is chunked, it sets response.body to a ReadableStream.

This example fetches a URL, treats the response body as a stream, and logs out the output until it's done streaming:

const response = await fetch(url);
const readableStream = response.body;
const reader = readableStream.getReader();
while (true) {
    const { done, value } = await;
    if (done) break;
    var text = new TextDecoder("utf-8").decode(value);
    console.log("Received ", text);

Streaming JSON

You might think it'd be super straightforward to stream JSON: just generate a JSON string on the server, and then JSON.parse the received text on the client. But there's a gotcha: the client could receive multiple JSON objects in the same chunk, and then an attempt to parse as JSON will fail.

The solution: JSON objects separated by new lines, known either as NDJSON or JSONlines.

This expression converts a Python dict to NDJSON, using the std lib json module:

json.dumps(some_dict) + "\n"

Here's how I actually used that, for one of the ChatGPT samples:"/chat")
def chat_handler():
    request_message = request.json["message"]

    def response_stream():
        response = openai.ChatCompletion.create(
            engine=os.getenv("AZURE_OPENAI_CHATGPT_DEPLOYMENT", "chatgpt"),
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": request_message},
        for event in response:
            yield json.dumps(event) + "\n"

    return Response(response_stream())

Consuming NDJSON streams in JavaScript

Once the server is outputting NDJSON, then we can write parsing code in JavaScript that splits by newlines and attempts to parse the resulting objects as JSON objects.

const response = await fetch(url);
const readableStream = response.body;
const reader = readableStream.getReader();
while (true) {
    const { done, value } = await;
    if (done) break;
    var text = new TextDecoder("utf-8").decode(value);
    const objects = text.split("\n");
    for (const obj of objects) {
        try {
            runningText += obj;
            let result = JSON.parse(runningText);
            console.log("Received", result);
            runningText = "";
        } catch (e) {
           // Not a valid JSON object

Since I need to use this same processing code in multiple Azure OpenAI samples, I packaged that into a tiny npm package called ndjson-readablestream.

Here's how you can use the package from JavaScript to make NDJSON parsing easier:

import readNDJSONStream from "ndjson-readablestream";

const response = await chatApi(request);
if (!response.body) {
    throw Error("No response body");
for await (const event of readNDJSONStream(response.body)) {
    console.log("Received", event);

For more examples of using the package, see this PR that uses it in a TypeScript component to render ChatGPT responses or usage in an HTML page, for a non-React ChatGPT sample.

I hope this helps other developers use NDJSON streams in your projects. Please let me know if you have suggestions for improving my approach!

Monday, August 7, 2023

Accessibility snapshot testing for Python web apps (Part 2)

In my previous post, I showed a technique that used axe-core along with pytest and Playwright to make sure that pages in your web apps have no accessibility violations. That's a great approach if it works for you, but realistically, most webpages have a non-zero number of accessibility violations, due to limited engineering resources or dependence on third-party libraries. Do we just give up on being able to test them? No! Instead, we use a different approach: snapshot testing.

Snapshot testing

Snapshot testing is a way to test that the output of a function matches a previously saved snapshot.

Here's an example using pytest-snapshot:

def emojify(s):
    return s.replace('love', '❤️').replace('python', '🐍')

def test_function_output_with_snapshot(snapshot):
    snapshot.assert_match(emojify('I love python'), 'snapshot.txt')

The first time we run the test, it will save the output to a file. We check the generated snapshots into source control. Then, the next time anyone runs that test, it will compare the output to the saved snapshot.

Snapshot testing + axe-core

First, a big kudos to Michael Wheeler from UMich and his talk on Automated Web Accessibility Testing for the idea of using snapshot testing with axe-core.

Here's the approach: We save snapshots of the axe-core violations and check them into source control. That way, our tests will let us know when new violations come up, and our snapshot files keep track of which parts of the codebase need accessibility improvements.

To make it as easy as possible, I made a pytest plugin that combines Playwright, axe-core, and snapshot testing.

python3 -m pip install pytest-axe-playwright-snapshot
python3 -m playwright install --with-deps

Here's an example test from a Flask app:

from flask import url_for
from playwright.sync_api import Page

def test_index(page: Page, axe_pytest_snapshot):
    page.goto(url_for("index", _external=True))

Running the snapshot tests

First run: We specify the --snapshot-update argument to tell the plugin to save the snapshots to file.

python3 -m pytest --snapshot-update

That saves a file like this one to a directory named after the test and browser engine, like snapshots/test_violations/chromium/snapshot.txt:

color-contrast (serious) : 2
empty-heading (minor) : 1
link-name (serious) : 1

Subsequent runs: The plugin compares the new snapshot to the saved snapshot, and asserts if they differ.

python3 -m pytest

Let's look through some example outputs next.

Test results

New accessibility issue 😱

If there are violations in the new snapshot that weren't in the old, the test will fail with a message like this:

E  AssertionError: New violations found: html-has-lang (serious)
E  That's bad news! 😱 Either fix the issue or run `pytest --snapshot-update` to update the snapshots.
E  html-has-lang - Ensures every HTML document has a lang attribute
E    URL:
E    Impact Level: serious
E    Tags: ['cat.language', 'wcag2a', 'wcag311', 'ACT']
E    Elements Affected:
E    1) Target: html
E       Snippet: <html>
E       Messages:
E       * The <html> element does not have a lang attribute

Fixed accessibility issue 🎉

If there are less violations in the new snapshot than the old one, the test will also fail, but with a happy message like this:

E  AssertionError: Old violations no longer found: html-has-lang (serious).
E  That's good news! 🎉 Run `pytest --snapshot-update` to update the snapshots.

CI/CD integration

Once you've got snapshot testing setup, it's a great idea to run it on every potential change to your codebase.

Here's an example of a failing GitHub action due to an accessibility violation, using this workflow file: Screenshot of GitHub actions workflow that shows test failure due to accessibility violations

Fixing accessibility issues

What should you do if you realize you've introduced an accessibility violation, or if you are tasked with reducing existing violations? You can read the reports from pytest to get a gist for the accessibility violations, but it's often easier to use a browser extension that uses the same Axe-core rules.

Also consider an IDE extension like VS Code Axe Linter

Don't rely on automation to find all issues

I think it's really important for web apps to measure their accessibility violations, so that they can avoid introducing accessibility regressions and eventually resolve existing violations. However, it's really important to note that these automated tools can only go so far. According to the axe-core docs, it finds about 57% of WCAG issues automatically. There can still be many issues with your site, like with tab order or keyboard access.

In addition to automation, please consider other ways to discover issues, such as paying for an external accessibility audit, engaging with your disabled users, and hiring engineers with disabilities.