AsyncEnvless is the asynchronous client. It has every method Envless has, with the same arguments and the same results, and every method that talks to the API is a coroutine you await. Both clients are built from one source, so they cannot drift apart: a fix to one is a fix to the other.
import asyncio
from envless import AsyncEnvless
async def main() -> None:
async with AsyncEnvless() as envless:
workspace = await envless.workspace.get()
async for project in envless.projects.iterate():
print(workspace['slug'], project['slug'])
asyncio.run(main())It runs on asyncio and on trio, through anyio, and reads ENVLESS_TOKEN and ENVLESS_API_ENDPOINT exactly as Envless does. Every method page shows both forms side by side.
What changes, and what does not
Envless | AsyncEnvless | |
|---|---|---|
| A call | envless.variables.list_all('api', 'production') | await envless.variables.list_all('api', 'production') |
| A stream | for variable in envless.variables.iterate(...) | async for variable in envless.variables.iterate(...) |
| The raw transport | envless.raw.request('/products') | await envless.raw.request('/products') |
| Closing | close() or with | await aclose() or async with |
| Its own HTTP client | http_client=httpx.Client(...) | http_client=httpx.AsyncClient(...) |
| Rotation | rotate_workspace_passphrase(client=envless, ...) | await rotate_workspace_passphrase(client=envless, ...) |
The iterate methods, iterate itself and every iterate_* such as iterate_history or products.access.iterate, are the ones you do not await: each returns an async iterator straight away, and requests the first page when the loop pulls the first item. Errors, retries, timeouts and results are identical in both clients, down to the bytes on the wire.
The ready made envless client from from envless import envless is the synchronous one. Async code builds its own AsyncEnvless, ideally once per event loop, and closes it when the loop is done with it. An httpx.AsyncClient belongs to the loop that opened its connections, so do not share one AsyncEnvless between loops or threads.
Run calls concurrently
Independent reads are where the async client earns its keep. Gather them and they share one connection pool:
import asyncio
from envless import AsyncEnvless
async def main() -> None:
async with AsyncEnvless() as envless:
environments = await envless.environments.list_all('api')
listings = await asyncio.gather(
*(envless.variables.list_all('api', environment['slug']) for environment in environments)
)
for environment, variables in zip(environments, listings, strict=True):
print(environment['slug'], len(variables))
asyncio.run(main())Concurrency does not make writes independent. Two variables.apply calls against the same environment still race on its baseline, and the loser fails with 409 variable.baseline_drifted. Run writes to one environment one at a time.
Cancellation and deadlines
timeout bounds each attempt, as on the synchronous client. On top of that, cancelling the task stops a call cleanly, retries and backoff included, and releases its connection. Any cancel scope works: asyncio.wait_for, asyncio.timeout on Python 3.11 and newer, trio's cancel scopes, or anyio:
import anyio
from envless import AsyncEnvless
async def main() -> None:
async with AsyncEnvless() as envless:
with anyio.move_on_after(5) as scope:
products = await envless.products.list_all()
print(len(products))
if scope.cancel_called:
print('gave up after 5 seconds')
anyio.run(main)A deadline you set yourself raises your runtime's cancellation, not EnvlessNetworkError. Only the client's own timeout turns into EnvlessNetworkError with is_timeout set.
In a FastAPI app
Open one client when the app starts and close it when the app stops, so every request shares the connection pool:
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from envless import AsyncEnvless
clients: dict[str, AsyncEnvless] = {}
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
async with AsyncEnvless() as envless:
clients['envless'] = envless
yield
app = FastAPI(lifespan=lifespan)
@app.get('/projects')
async def projects() -> list[str]:
page = await clients['envless'].projects.list()
return [project['slug'] for project in page['items']]Keep key derivation off the loop
The crypto helpers are plain synchronous functions, on purpose: encryption is CPU work, not I/O. encrypt_with_key and decrypt_with_key take microseconds, but deriving a key from a passphrase runs 200,000 rounds of PBKDF2, and encrypt_value and decrypt_value derive on every call. In async code, derive the key once, in a worker thread, and use the _with_key functions from then on:
import asyncio
import os
from envless import AsyncEnvless, decrypt_with_key, derive_workspace_key
async def main() -> None:
async with AsyncEnvless() as envless:
identity = await envless.me.get()
key = await asyncio.to_thread(
derive_workspace_key, os.environ['ENVLESS_PASSPHRASE'], identity['workspaceId']
)
variables = await envless.variables.list_all('api', 'production')
plaintext = {
variable['name']: decrypt_with_key(key, variable['value'])
for variable in variables
if variable['value'] is not None
}
print(sorted(plaintext))
asyncio.run(main())anyio.to_thread.run_sync does the same on trio. A WorkspaceKey is safe to share between threads and tasks.
Which one to use
| Your code | Use |
|---|---|
| A script, a CLI, a cron job, a Django view, a Celery task | Envless, or the ready made envless |
| FastAPI, Starlette, aiohttp, Quart, an async worker | AsyncEnvless |
| Many independent reads that should overlap | AsyncEnvless with asyncio.gather or a task group |
| Threads rather than an event loop | One Envless, shared between the threads |