Python async batch lab: bounded concurrency with a result for every input

Record a repeatable input set

Download the Python file and begin with a handful of short, distinct prompts. Give each input a stable identifier. Record the Python version, selected model ID and concurrency setting so another run can be compared meaningfully. Keep the API key in the environment, not in the saved input or result file.

The example uses urllib for HTTP and asyncio.to_thread to move blocking work away from the event loop. A semaphore surrounds each complete request operation. This limits concurrent work; it does not create a requests-per-minute quota or guarantee that shared provider capacity will accept every request.

Companion Python example

Download batch.py

"""Python 3.11+. Each item makes one paid call. No automatic retries."""
import asyncio
import json
import os
import urllib.request
from urllib.error import HTTPError, URLError
from http.client import HTTPException

def request_one(prompt, key, model, opener=urllib.request.urlopen):
    request=urllib.request.Request(
        'https://uncensuredai.com/v1/chat/completions',
        data=json.dumps({'model':model,'messages':[{'role':'user','content':prompt}],'stream':False}).encode(),
        headers={'Authorization':f'Bearer {key}','Content-Type':'application/json'},
    )
    with opener(request, timeout=90) as response:
        result=json.load(response)
    return result['choices'][0]['message']['content']

async def batch(prompts, key, model='glm-5.3-flash-abliterated', concurrency=2, sender=request_one):
    if not key:
        raise ValueError('Set UNCENSUREDAI_API_KEY in the server environment.')
    if isinstance(concurrency,bool) or not isinstance(concurrency,int) or concurrency<1:
        raise ValueError('Concurrency must be a positive integer.')
    semaphore=asyncio.Semaphore(concurrency)
    async def run(index,prompt):
        async with semaphore:
            try:
                text=await asyncio.to_thread(sender,prompt,key,model)
                return {'index':index,'ok':True,'text':text}
            except HTTPError as error:
                return {'index':index,'ok':False,'error':f'HTTP {error.code}; no automatic retry'}
            except (URLError,OSError,HTTPException):
                return {'index':index,'ok':False,'error':'Network failure or timeout; outcome may be unknown; no automatic retry'}
            except (KeyError,IndexError,TypeError,ValueError):
                return {'index':index,'ok':False,'error':'Unexpected response; inspect locally without logging credentials'}
    return await asyncio.gather(*(run(i,p) for i,p in enumerate(prompts)))

if __name__=='__main__':
    results=asyncio.run(batch(['Summarize what an API is.','Explain input versus output tokens.'],os.environ.get('UNCENSUREDAI_API_KEY'),model=os.environ.get('MODEL','glm-5.3-flash-abliterated')))
    # Prints model outputs locally. Choose your own retention policy before using real customer prompts.
    print(json.dumps(results,indent=2))

Verify the batch contract locally

Use a local fixture endpoint or an injected fake HTTP function with successful responses, delayed responses, malformed JSON and HTTP errors. Run the same inputs sequentially, then with a small concurrency limit. Assert that each input appears exactly once in the results and that its result stays associated with the correct identifier.

Instrument the fixture to count active operations. Its peak should respect the chosen limit during an ordinary completed run. Measure that behavior directly rather than assuming that a semaphore placed anywhere in the program necessarily bounds the network operations you intended.

Treat timeouts as outcomes, not invitations to replay

Configure a timeout for blocking HTTP operations and record failures per input. urllib’s timeout is not a promise of a strict total wall-clock deadline. Cancelling an awaiting coroutine also does not reliably stop a blocking operation already running in a worker thread.

Use one attempt per input in this lab. Do not automatically resubmit a timed-out item: the server may have received or processed it. Keep the unknown outcome visible, and separate any later operator-chosen rerun from the original batch. This makes request counts and possible spending easier to reconcile.

Keep local evidence separate from paid execution

Fixture results verify your scheduling, association and error paths. They do not establish live model quality or throughput. A run against the real API makes multiple requests and can consume paid credit; concurrency changes overlap, not the fact that each request may incur cost.

Start with the downloaded file and a tiny workload. Confirm the connection settings, select a model from the current catalog and compare completed outcomes before increasing volume. Keep prompts out of diagnostic logs when they are unnecessary for the experiment.