Request middleware

One place for policy, audit, and observability.

Middleware wraps each request with its parameters, an engine snapshot, and its result, so platform controls do not have to be repeated at every call site.

How it works

A middleware is a callable middleware(context, call_next). It can inspect or change context, call await call_next(context) to continue down the chain, and return the result — or return without calling call_next to short-circuit the request.

import logging
import time

from xbbg import blp

log = logging.getLogger('bloomberg.audit')

@blp.add_middleware
async def audit(context, call_next):
    context.metadata['desk'] = 'research'
    started = time.perf_counter()
    try:
        return await call_next(context)
    finally:
        log.info(
            '%s %s %s securities=%d fields=%d error=%s %.1fms',
            context.request_id,
            context.params.service,
            context.params.operation,
            len(context.securities),
            len(context.fields),
            type(context.error).__name__ if context.error else None,
            (time.perf_counter() - started) * 1000,
        )

Middleware runs in registration order: the first one registered is the outermost. Both async def and plain functions are accepted; a plain function can do work before the request and return call_next(context) (the awaitable is awaited for it), but only an async middleware can act on the result. Since 1.8.0 each request takes a snapshot of the chain when it starts, so adding or removing middleware does not affect requests already in flight.

What runs through middleware

Middleware wraps requests that go through request() / arequest():

• the request helpers bdp, bds, bdh, bdib, bdtick, bql, bsrch, beqs, blkp, bport, bqr, bcurves, bgovts, and bflds (sync and async), plus the bfld/fieldInfo/fieldSearch aliases;
• request() / arequest() themselves;
• xbbg.ext analytics built on those helpers: the bond_*, option_*, and cdx_* functions.

It does not run for subscriptions and streams, for xbbg.ext recipes implemented in the native engine (for example yas, preferreds, corporate_bonds, ext.bqr, futures and CDX resolution, dividends, earnings, turnover, ETF holdings and NAV, auctions, vol_surface, index_members), for bta, for delay_status and the entitlement helpers, or for schema and field-cache lookups.

The chain is process-wide: it applies to the global engine and to scoped xbbg.Engine(...) contexts alike.

RequestContext

Attribute	Contents
request_id	xbbg's label for this request (req-<nanoseconds>)
params	The RequestParams dataclass to be dispatched: service, operation, securities, fields, overrides, elements, dates, field_types, format, return_eids, on_delayed, ... Changes made before call_next are what Bloomberg receives. Since 1.8.0, params.field_types holds only the caller's overrides while the chain runs; the engine resolves the remaining types when the request is sent
field_type_default	Default type the request terminal applies to fields without type information ('string' for bdp, 'float64' for bdh, otherwise None)
securities, fields	Normalized lists taken from the request
backend	The requested backend (None for the typed helpers, which convert after the chain)
raw	Whether the caller asked for the raw Arrow result
environment	Read-only RequestEnvironment snapshot (below)
metadata	Free-form dict shared by every middleware on this request
started_at	time.perf_counter() value when the context was created
elapsed_ms	Bloomberg round-trip time, set after the request completes or fails
batch, table	The engine's native result (table is an ArrowTable) once the request succeeds
frame	The converted result, when conversion has happened
error	The exception, if the request or a middleware failed

call_next(context) returns the engine result — normally the native ArrowTable; conversion to the caller's backend happens after the chain.

RequestEnvironment

context.environment describes the engine serving the request, without credentials:

Attribute	Contents
source	'default_engine' (never configured), 'global_config' (after configure()), or 'scoped_engine' (inside xbbg.Engine(...))
host, port, servers	Direct connection settings
zfp_remote	ZFP leased-line remote, if any
auth_method, app_name, user_id	Authentication identity settings
validation_mode	Field validation mode

Examples

Enforce a request policy

from xbbg import blp

ALLOWED_SERVICES = {'//blp/refdata', '//blp/apiflds'}

@blp.add_middleware
def enforce_policy(context, call_next):
    if context.params.service not in ALLOWED_SERVICES:
        raise PermissionError(f'{context.params.service} is not approved for this application')
    if len(context.securities) > 500:
        raise ValueError('split requests larger than 500 securities')
    if context.environment.source == 'default_engine':
        context.metadata['warning'] = 'running against the default local engine'
    return call_next(context)

Cache reference data in-process

from xbbg import blp

_cache = {}

@blp.add_middleware
async def cache_reference_data(context, call_next):
    params = context.params
    if params.operation != 'ReferenceDataRequest':
        return await call_next(context)
    key = (
        params.service,
        tuple(context.securities),
        tuple(context.fields),
        tuple(params.overrides or ()),
        params.format,
    )
    if key in _cache:
        context.metadata['cache'] = 'hit'
        return _cache[key]
    result = await call_next(context)
    _cache[key] = context.table
    return result

A short-circuit result that is Arrow data (such as a cached ArrowTable) is converted to the caller's backend as usual; any other value is returned to the caller unchanged.

Count errors by class

from collections import Counter

from xbbg import blp
from xbbg.exceptions import BlpError

error_counts = Counter()

@blp.add_middleware
async def count_errors(context, call_next):
    try:
        return await call_next(context)
    except BlpError as exc:
        error_counts[type(exc).__name__] += 1
        raise

Managing the chain

Function	Effect
add_middleware(fn)	Append fn; returns fn, so it works as a decorator
remove_middleware(fn)	Remove a registered middleware
clear_middleware()	Remove all middleware
get_middleware()	Current chain as a tuple
set_middleware([fn, ...])	Replace the chain

All are importable from xbbg and xbbg.blp; RequestContext and RequestEnvironment are importable from both as well.

Testing middleware without Bloomberg

xbbg.testing.mock_engine is itself a middleware that serves canned responses. Register your middleware first so it runs outside the mock. Since 1.8.0, field types are resolved in the request terminal, so a middleware that answers a request — the mock or a cache — never starts an engine: bdp(), bdh(), and request() all run offline under mock_engine. (In 1.7.0 and earlier, bdp() and bdh() looked up field types first and needed a Bloomberg session; request() did not.)

from xbbg import blp
from xbbg.testing import create_mock_response, mock_engine

calls = []

@blp.add_middleware
async def record(context, call_next):
    calls.append((context.params.operation, list(context.securities)))
    return await call_next(context)

response = create_mock_response(
    service='//blp/refdata',
    operation='ReferenceDataRequest',
    data={'AAPL US Equity': {'PX_LAST': 101.23}},
)

with mock_engine([response]):
    df = blp.request(
        '//blp/refdata',
        'ReferenceDataRequest',
        securities=['AAPL US Equity'],
        fields=['PX_LAST'],
        backend='pandas',
    )

assert calls == [('ReferenceDataRequest', ['AAPL US Equity'])]
blp.remove_middleware(record)

See Testing for more.

Related

• Configuration — engine settings reflected in RequestEnvironment
• Errors and Warnings — exception classes middleware can observe
• Troubleshooting — logging and diagnostics