/Docs

FastAPI integration

This guide covers FastAPI (ASGI) and Flask (WSGI) — both use the same avsb-python package with different middleware shims. FastAPI users get full async support via AsyncAvsbServer; Flask users use the synchronous AvsbServer with the WSGI shim. By the end your app will initialise the SDK once on startup and expose a per-request flag helper via request.state.avsb (FastAPI) or g.avsb (Flask).

FastAPI (ASGI)

1

Install

terminal
bash
1pip install avsb-python
2

Obtain your SDK key

Open your A vs B project, go to Settings → Environments, and copy the Server SDK key. Store it in an environment variable — never hardcode it.

.env
bash
1AVSB_SDK_KEY=sdk_production_...
3

Initialise the SDK and add ASGI middleware

Use FastAPI's lifespan context to start and stop the SDK cleanly. Add the ASGI middleware so every request gets a scoped helper attached to request.state.avsb.

main.py
python
1import os
2from contextlib import asynccontextmanager
3from typing import AsyncIterator
4
5from fastapi import FastAPI, Request
6from fastapi.responses import JSONResponse
7from avsb import AsyncAvsbServer
8from avsb.contrib.asgi import AvsbMiddleware
9
10server: AsyncAvsbServer | None = None
11
12
13@asynccontextmanager
14async def lifespan(app: FastAPI) -> AsyncIterator[None]:
15 global server
16 server = AsyncAvsbServer(sdk_key=os.environ['AVSB_SDK_KEY'])
17 await server.on_ready()
18 yield
19 await server.close()
20
21
22app = FastAPI(lifespan=lifespan)
23
24app.add_middleware(
25 AvsbMiddleware,
26 server_factory=lambda: server,
27 context_from=lambda request: {
28 'kind': 'user',
29 'key': request.headers.get('x-user-id', 'anon'),
30 'plan': request.headers.get('x-user-plan', 'free'),
31 },
32)
4

Read a flag in a route

main.py
python
1@app.get('/checkout')
2async def checkout(request: Request):
3 avsb = request.state.avsb # pre-scoped to the current user context
4 show_new = avsb.get_bool_flag('checkout-v2', False)
5 theme = avsb.get_string_flag('ui-theme', 'default')
6 return JSONResponse({'showNewCheckout': show_new, 'theme': theme})
5

Track an event

python
1request.state.avsb.track('purchase', value=149.99, properties={'currency': 'usd'})
6

Identify a user

To re-scope to a different user context mid-request (for example after resolving a JWT to a real user record):

python
1scoped = server.for_user({
2 'kind': 'user',
3 'key': str(user.id),
4 'plan': user.plan,
5 'org_id': str(user.org_id),
6})
7show_new = await scoped.get_bool_flag('checkout-v2', False)

Flask (WSGI)

Flask runs on the WSGI protocol. The pattern is the same but uses the synchronousAvsbServerand the WSGI middleware shim. The scoped helper is available via Flask's application context global g.avsb.

app.py
python
1import os
2from flask import Flask, g, jsonify, request as flask_request
3from avsb import AvsbServer
4from avsb.contrib.wsgi import AvsbFlaskExtension
5
6app = Flask(__name__)
7
8avsb_ext = AvsbFlaskExtension(
9 sdk_key=os.environ['AVSB_SDK_KEY'],
10 context_from=lambda: {
11 'kind': 'user',
12 'key': flask_request.headers.get('x-user-id', 'anon'),
13 'plan': flask_request.headers.get('x-user-plan', 'free'),
14 },
15)
16avsb_ext.init_app(app)
17
18
19@app.route('/checkout')
20def checkout():
21 show_new = g.avsb.get_bool_flag('checkout-v2', False)
22 return jsonify({'showNewCheckout': show_new})
Flask vs FastAPI middleware shape
Flask uses the AvsbFlaskExtension helper (which wraps the WSGI middleware from avsb.contrib.wsgi) rather than app.add_middleware. Conceptually they do the same thing: bootstrap once, scope per request, attach to request context.

Graceful shutdown

For FastAPI the lifespan handler already calls await server.close() on shutdown. For Flask, register an atexit hook:

app.py
python
1import atexit
2atexit.register(avsb_ext.server.close_sync)

Testing

tests/test_routes.py
python
1import pytest
2from avsb.test_utils import MockAsyncAvsbServer, TestData
3from httpx import AsyncClient
4
5
6@pytest.fixture
7def mock_server():
8 td = (
9 TestData.flag('checkout-v2')
10 .boolean_flag()
11 .variation_for_user('u_power', True)
12 .fallthrough_variation(False)
13 )
14 return MockAsyncAvsbServer(flags=[td.build()])
15
16
17@pytest.mark.anyio
18async def test_checkout_flag(mock_server, monkeypatch):
19 monkeypatch.setattr('main.server', mock_server)
20 async with AsyncClient(app=app, base_url='http://test') as client:
21 resp = await client.get('/checkout', headers={'x-user-id': 'u_power'})
22 assert resp.json()['showNewCheckout'] is True

What's next