This document covers the full public API exported by from turbo import * in py-turbo-api.
The package exports these symbols:
Turbo, APIRouter, TurboSettingsRequest, WebSocket, UploadFile, ConnectionManager, normalize_ws_close_code, ws_close_reasonResponse, JSONResponse, TextResponse, HTMLResponse, RedirectResponse, StreamingResponse, EventSourceResponse, SSEEvent, encode_sse_event, negotiate_content_type, NegotiatedResponse, build_cache_control, with_cache_headers, FileResponse, BackgroundTask, register_json_encoderHTTPErrorDepends, Security, ClassDepends, DependencyGroup, dependency_group, Query, Header, Cookie, Form, File, Host, Bodyapp_state_dependency, get_app_stateModel, field, field_validator, model_validator, type_validatorapi_key_auth, bearer_auth, jwt_auth, oauth2_bearer, oauth2_authorization_code, oauth2_client_credentials, csrf_token, csrf_protect, JWKSCache, websocket_token_auth, websocket_jwt_authCORSMiddleware, GZipMiddleware, CompressionMiddleware, RateLimitMiddleware, ResponseCacheMiddleware, TrustedHostMiddleware, SessionMiddleware, CSRFMiddleware, HTTPSRedirectMiddleware, ProxyHeadersMiddleware, MemorySessionBackendRequestIDMiddleware, StructuredLoggingMiddleware, MetricsMiddleware, PrometheusMiddleware, TracingMiddleware, OpenTelemetryTracingHook, LogEvent, MetricEvent, get_request_id, set_request_idTestClient, AsyncTestClient, WebSocketTestSession, TestResponseInMemoryJobQueue, RetryPolicy, JobRecord, CeleryQueueAdapter, RQQueueAdapter, RedisQueueAdapterfrom turbo import Turbo, Model, field, Query, Depends, HTTPError
app = Turbo(title="Todo API", version="1.0.0")
class TodoIn(Model):
title: str = field(min_len=1, max_len=200)
class TodoOut(Model):
id: int
title: str
def require_tenant(x_tenant: str = Query(alias="tenant")):
if not x_tenant:
raise HTTPError(400, "tenant is required")
return x_tenant
@app.post("/todos", response_model=TodoOut, status_code=201)
async def create_todo(body: TodoIn, tenant=Depends(require_tenant)):
return {"id": 1, "title": body.title}
TurboCreate the ASGI app and register routes, middleware, docs, and lifecycle handlers.
Constructor:
Turbo(
*,
request_timeout=10.0,
max_body_bytes=1_000_000,
max_concurrency=200,
title="TurboAPI",
version="0.1.0",
multipart_max_fields=1000,
multipart_max_file_size=10_000_000,
multipart_spool_threshold=1_000_000,
multipart_max_part_size=10_000_000,
redirect_slashes=True,
redirect_status_code=307,
openapi_url="/openapi.json",
docs_url="/docs",
redoc_url="/redoc",
swagger_js_url="https://unpkg.com/swagger-ui-dist@5/swagger-ui-bundle.js",
swagger_css_url="https://unpkg.com/swagger-ui-dist@5/swagger-ui.css",
redoc_js_url="https://unpkg.com/redoc@2/bundles/redoc.standalone.js",
docs_auth=None,
operation_id_strategy="function",
operation_id_generator=None,
shutdown_drain_timeout=10.0,
dependencies=None,
)
Main methods:
route, get, post, put, delete, patch, head, options, add_api_route, websocketinclude_routeruse (request/response middleware), use_asgi (ASGI middleware)mount_static, mount, mount_hostadd_state_resource, remove_state_resource, state_dependencyset_openapi_transform, openapi_transform, set_openapi_extension, remove_openapi_extension, set_operation_id_generator, set_openapi_servers, add_openapi_server, clear_openapi_servers, set_openapi_security, add_openapi_security_requirement, clear_openapi_security, set_openapi_reuse_parameters, enable_docs_self_hostset_docs_auth, docs_authon_event, startup, shutdownexception_handleroverride_dependency, override_dependencies, override_scope, clear_dependency_overridesdependency_graph, dependency_graph_for_route, format_dependency_graphuse_extension, get_extension, register_auth_provider, get_auth_provider, register_telemetry_exporter, get_telemetry_exporter, register_cache_backend, get_cache_backendjson_encoderTurbo.from_settings(...), Turbo.from_env(prefix="TURBO_")Route decorator extras are available on HTTP methods and route(...):
name, operation_idresponse_model, status_codeinclude_in_schema, internaltags, summary, descriptionresponses, security, deprecatedcallbacks, webhooks, examplesresponse_description, openapi_extradependencies (list of Depends(...) or callables, executed before handler)WebSocket decorator extras on websocket(...):
name, operation_idinclude_in_schematags, summary, descriptiondeprecated, examples, openapi_extrasubprotocolsdependencies (list of Depends(...) or callables, executed before handler)APIRouterSub-router for route grouping and composition.
APIRouter(*, prefix="", tags=None, dependencies=None)
dependencies applies shared dependencies to all routes in the router.
Methods:
route, get, post, put, delete, patch, head, optionsadd_api_routeinclude_router(router, prefix="", tags=None, dependencies=None) can add extra dependencies to all included routes.
Usage:
from turbo import Turbo, APIRouter
app = Turbo()
users = APIRouter(prefix="/users", tags=["users"])
@users.get("/{user_id:int}")
async def get_user(user_id: int):
return {"id": user_id}
app.include_router(users, prefix="/v1")
TurboSettingsDataclass for env-driven runtime config.
TurboSettings(
request_timeout=10.0,
max_body_bytes=1_000_000,
max_concurrency=200,
multipart_max_fields=1000,
multipart_max_file_size=10_000_000,
multipart_spool_threshold=1_000_000,
multipart_max_part_size=10_000_000,
redirect_slashes=True,
redirect_status_code=307,
openapi_url="/openapi.json",
docs_url="/docs",
redoc_url="/redoc",
shutdown_drain_timeout=10.0,
title="TurboAPI",
version="0.1.0",
)
Methods:
TurboSettings.from_env(prefix="TURBO_")to_turbo_kwargs()RequestHTTP request wrapper.
Key properties:
method, pathappstatepath_paramsheaders, headers_multiquery_params, query_params_multicookiesrequest_idcsrf_tokensessionMethods:
stream(receive)body(receive)json(receive)form_multi(receive)form(receive)parse_payload(receive, media_type=None)set_session(data)set_session_value(key, value)clear_session()Notes:
request.app references the current Turbo application instance.request.state is request-scoped mutable state.app.state is process-scoped mutable state shared across requests.app.add_state_resource(name, factory, cleanup=None) registers startup bootstrapping + shutdown cleanup for app.state.<name>.app.remove_state_resource(name) removes a registered resource.app.state_dependency(name, default=..., required=True) creates a dependency that reads app.state.app_state_dependency(name, default=..., required=True, expected_type=None) helper version that returns Depends(...).get_app_state(request, name, default=..., expected_type=None) typed getter.UploadFileUploaded multipart file object.
filename, content_type, file, sizeread(), seek(offset), close()spooled_to_diskWebSocketWebSocket connection wrapper with helpers.
Key properties:
path, path_paramsheaders, headers_multiquery_params, query_params_multicookiesrequested_subprotocolsaccepted, closedclose_code, close_reasonidle_secondsMethods:
accept(subprotocol=None), accept_subprotocol(allowed, fallback=None), select_subprotocol(allowed)receive(), receive_text(), receive_json(), receive_with_idle_timeout(timeout, close_code=1001, reason="Idle timeout")send_text(text), send_bytes(data), send_json(data), send_ping(payload="turbo:ping"), send_pong(payload="turbo:pong")touch(), start_heartbeat(...), close(code=1000, reason=None), close_with_reason(...)ConnectionManagerTracks active sockets and group membership.
Methods:
connect, add, remove, disconnectjoin, leave, list_groupssend_text, send_json, broadcast_text, broadcast_jsonProperty:
active_countnormalize_ws_close_code(code)ws_close_reason(code)Response(status=200, headers=None, body=b"", background=None)JSONResponse(data, status=200, headers=None, dumps=None, encoders=None, background=None)TextResponse(text, status=200, headers=None, background=None)HTMLResponse(html, status=200, headers=None, background=None)RedirectResponse(url, status=307, headers=None, background=None)StreamingResponse(content, status=200, headers=None, media_type="application/octet-stream", background=None)EventSourceResponse(events, status=200, headers=None, ping_interval=15.0, ping_message="ping", background=None)FileResponse(path, status=200, headers=None, filename=None, chunk_size=65536, background=None)NegotiatedResponse(accept_header, variants, status=200, headers=None, default_media_type=None, background=None)SSEEvent(data, event=None, id=None, retry=None, comment=None)encode_sse_event(event)negotiate_content_type(accept_header, available, default=None)build_cache_control(...)with_cache_headers(headers=None, cache_control=None, etag=None, last_modified=None)BackgroundTask(fn, *args, **kwargs)register_json_encoder(type_, encoder_fn)app.json_encoder(type_, encoder_fn)HTTPErrorHTTPError(status, message="Error", detail=None)
Raised to return structured error responses (for example raise HTTPError(404, "Not Found")).
Depends(call, cache=True, scopes=None)Security(call, scopes=None, cache=True)ClassDepends(cls, cache=True) for class-based dependency constructors.dependency_group(*items) returns DependencyGroup(...) for reusable dependency sets.Query(alias=None, required=True, description=None, example=None, examples=None, deprecated=False, schema=None)Header(alias=None, required=True, description=None, example=None, examples=None, deprecated=False, schema=None)Cookie(alias=None, required=True, description=None, example=None, examples=None, deprecated=False, schema=None)Form(alias=None, required=True, description=None, example=None, examples=None, deprecated=False, schema=None)File(alias=None, required=True, description=None, example=None, examples=None, deprecated=False, schema=None)Host(alias=None, required=True, description=None, example=None, examples=None, deprecated=False, schema=None)Body(alias=None, required=True, media_type=None, embed=None, description=None, example=None, examples=None, schema=None)Example:
from turbo import Turbo, Query, Header, Depends
app = Turbo()
def auth(authorization: str = Header(alias="authorization")):
return authorization
@app.get("/items/{item_id:int}")
async def read_item(
item_id: int,
q: str = Query(required=False),
token=Depends(auth),
):
return {"item_id": item_id, "q": q, "token": token}
ModelTyped validation and schema base class.
Optional compatibility:
pydantic>=2 is installed, pydantic.BaseModel can also be used for request bodies and response_model.field(min_len=None, max_len=None, ge=None, le=None, gt=None, lt=None, multiple_of=None, min_items=None, max_items=None, regex=None, discriminator=None, alias=None)field_validator(*field_names, mode="after")model_validator(mode="after")type_validator(type_, mode="after") for non-Model custom types.Example:
from turbo import Model, field, field_validator, model_validator
class UserIn(Model):
email: str = field(min_len=5, max_len=320)
age: int = field(ge=13)
@field_validator("email")
def normalize_email(cls, value: str):
return value.strip().lower()
@model_validator()
def ensure_domain(cls, data):
if "@example.com" not in data["email"]:
raise ValueError("email domain must be @example.com")
return data
api_key_auth(header_name="x-api-key", scheme_name="ApiKeyAuth", auto_error=True)bearer_auth(scheme_name="BearerAuth", bearer_format=None, auto_error=True)jwt_auth(secret=None, scheme_name="JWTAuth", issuer=None, audience=None, leeway=0, auto_error=True, algorithms=None, public_key=None, jwks_url=None, jwks_cache=None)oauth2_bearer(token_url, scheme_name="OAuth2", scopes=None, secret=None, public_key=None, jwks_url=None, algorithms=None, issuer=None, audience=None, leeway=0, auto_error=True)oauth2_authorization_code(authorization_url, token_url, refresh_url=None, scheme_name="OAuth2AuthorizationCode", scopes=None, secret=None, public_key=None, jwks_url=None, algorithms=None, issuer=None, audience=None, leeway=0, auto_error=True)oauth2_client_credentials(token_url, refresh_url=None, scheme_name="OAuth2ClientCredentials", scopes=None, secret=None, public_key=None, jwks_url=None, algorithms=None, issuer=None, audience=None, leeway=0, auto_error=True)csrf_token(req, auto_error=True)csrf_protect(header_name="x-csrf-token", auto_error=True)websocket_token_auth(query_param="token", scheme_name="WebSocketTokenAuth", auto_error=True)websocket_jwt_auth(secret=None, query_param="token", scheme_name="WebSocketJWTAuth", issuer=None, audience=None, leeway=0, auto_error=True, algorithms=None, public_key=None, jwks_url=None)JWKSCache(ttl_seconds=300, fetcher=None)get(url)Add middleware via app.use(...) or app.use_asgi(...).
CORSMiddleware(allow_origins=None, allow_methods=None, allow_headers=None, expose_headers=None, allow_credentials=False, max_age=600, allow_origin_regex=None)GZipMiddleware(minimum_size=500)CompressionMiddleware(minimum_size=500, prefer=None, brotli_quality=5, gzip_level=6, deflate_level=6)RateLimitMiddleware(max_requests=60, window_seconds=60, key_fn=None, include_headers=True)ResponseCacheMiddleware(ttl_seconds=5, max_entries=512, methods=None, cache_statuses=None, key_fn=None, vary_headers=None)TrustedHostMiddleware(allowed_hosts)SessionMiddleware(secret_key, cookie_name="session", max_age=1209600, same_site="Lax", https_only=False, path="/", domain=None, http_only=True, partitioned=False, signer_salt="turbo.session", signer_digest="sha256", secret_key_fallbacks=None, backend=None, session_id_bytes=24)CSRFMiddleware(cookie_name="csrftoken", header_name="x-csrf-token", safe_methods=None, exempt_paths=None, use_session=True, session_key="csrf_token", same_site="Lax", https_only=False, path="/", domain=None)HTTPSRedirectMiddleware(redirect_status_code=307)ProxyHeadersMiddleware(trusted_hosts=None, trusted_cidrs=None, forwarded_proto_header="x-forwarded-proto", forwarded_for_header="x-forwarded-for", forwarded_host_header="x-forwarded-host")MemorySessionBackend() with methods get, set, deleteRequestIDMiddleware(header_name="x-request-id", response_header_name="x-request-id", generator=None)StructuredLoggingMiddleware(hook)MetricsMiddleware(hooks)PrometheusMiddleware(endpoint="/metrics", duration_buckets=None, include_process_label=True, multiprocess_dir=None, aggregate_workers=True)TracingMiddleware(hook)OpenTelemetryTracingHook(tracer=None, name="turbo.request")Events:
LogEvent(scope_type, method, path, route, status_code, duration_ms, request_id, error=None)MetricEvent(scope_type, method, path, route, status_code, duration_ms, request_id)Helpers:
get_request_id()set_request_id(request_id)TestClientSynchronous test client for app-level testing.
Constructor:
TestClient(app)
Request methods:
request(method, path, headers=None, params=None, json_body=None, data=None, content=None)get, post, put, patch, delete, head, optionsdependency_override(original, override) context managerdependency_overrides({original: override, ...}) context manageroverride_scope(name="scope") context managerTestResponsestatus_code, headers, contentjson()Example:
from turbo import Turbo, TestClient
app = Turbo()
@app.get("/ping")
async def ping():
return {"ok": True}
def test_ping():
client = TestClient(app)
resp = client.get("/ping")
assert resp.status_code == 200
assert resp.json() == {"ok": True}
AsyncTestClientAsync test client with lifespan support and HTTP/WebSocket helpers.
async with AsyncTestClient(app) as client: ...request, get, post, put, patch, delete, head, optionsawait client.websocket_connect(path, headers=None, params=None, subprotocols=None)dependency_override(...), dependency_overrides(...), override_scope(...)WebSocketTestSessionReturned by AsyncTestClient.websocket_connect(...).
send_text, send_bytes, send_jsonreceive, receive_text, receive_jsoncloseaccepted, closed, subprotocolInMemoryJobQueueIn-process async queue for background jobs.
register(name, handler)start(workers=1), stop(), join(timeout=None)enqueue(name, payload=None, delay_seconds=None, run_at=None, retry=None, idempotency_key=None)get_job(job_id)RetryPolicy(max_retries=0, base_delay=1.0, backoff=2.0, max_delay=60.0, jitter=0.0)JobRecord(id, name, payload, run_at, status, attempts, result, error, idempotency_key, ...)CeleryQueueAdapter(celery_app)RQQueueAdapter(rq_queue)RedisQueueAdapter(redis_client, list_name="turbo:jobs")
turbo.extensions)TurboExtension protocol (name, setup(app))ExtensionRegistry(auth_providers, telemetry_exporters, cache_backends)register_extension_hook(app, hook)run_extension_hooks(app, event, **kwargs)setup_extension(extension, app)turbo.integrations)create_sqlalchemy_engine(url, **kwargs)register_sqlalchemy(app, url, ...)make_sqlalchemy_session_dependency(...)AuthContextbuild_bearer_guard(...)build_scope_guard(auth_dep, required_scopes=[...])PageParamsparse_pagination(...)apply_pagination(items, params)apply_sorting(items, sort=..., order=...)apply_filters(items, filters)load_pydantic_settings(SettingsCls, **kwargs)settings_dependency(SettingsCls, cache=True, **kwargs)Use /docs as the Pages source:
main (or master) and folder /docs.docs/index.md as landing page and link all sections from there.