Schema Introspection

xbbg.schema

Bloomberg schema introspection and stub generation.

This module provides access to cached Bloomberg service schemas and can generate Python type stubs for IDE autocomplete support.

Example:

import xbbg from xbbg.schema import get_schema, list_operations

Get schema for a service

schema = await get_schema(“//blp/refdata”)

List available operations

ops = await list_operations(“//blp/refdata”) print(ops) # [‘ReferenceDataRequest’, ‘HistoricalDataRequest’, …]

Get enum values for an element

values = await get_enum_values(“//blp/refdata”, “ReferenceDataRequest”, “periodicitySelection”)

ElementInfo Objects

@dataclass
class ElementInfo()

Schema element information.

from_dict

@classmethod
def from_dict(cls, d: dict[str, Any]) -> ElementInfo

Create from dictionary (parsed JSON).

OperationSchema Objects

@dataclass
class OperationSchema()

Schema for a service operation.

from_dict

@classmethod
def from_dict(cls, d: dict[str, Any]) -> OperationSchema

Create from dictionary (parsed JSON).

ServiceSchema Objects

@dataclass
class ServiceSchema()

Schema for a Bloomberg service.

from_dict

@classmethod
def from_dict(cls, d: dict[str, Any]) -> ServiceSchema

Create from dictionary (parsed JSON).

from_json

@classmethod
def from_json(cls, json_str: str) -> ServiceSchema

Create from JSON string.

get_operation

def get_operation(name: str) -> OperationSchema | None

Get an operation by name.

aget_schema

async def aget_schema(service: str) -> ServiceSchema

Get schema for a service (async).

Loads from cache if available, otherwise introspects the service.

Arguments:

• service - Service URI (e.g., “//blp/refdata”)

Returns:

ServiceSchema object with operations and element definitions.

aget_operation

async def aget_operation(service: str, operation: str) -> OperationSchema

Get schema for a specific operation (async).

Arguments:

• service - Service URI (e.g., “//blp/refdata”)
• operation - Operation name (e.g., “ReferenceDataRequest”)

Returns:

OperationSchema object with request/response definitions.

alist_operations

async def alist_operations(service: str) -> list[str]

List all operations for a service (async).

Arguments:

• service - Service URI (e.g., “//blp/refdata”)

Returns:

List of operation names.

aget_enum_values

async def aget_enum_values(service: str, operation: str,
                           element: str) -> list[str] | None

Get valid enum values for an element (async).

Arguments:

• service - Service URI
• operation - Operation name
• element - Element name

Returns:

List of valid enum values, or None if not an enum.

alist_valid_elements

async def alist_valid_elements(service: str,
                               operation: str) -> list[str] | None

List all valid element names for an operation (async).

Arguments:

• service - Service URI
• operation - Operation name

Returns:

List of valid element names.

get_schema

def get_schema(service: str) -> ServiceSchema

Get schema for a service (sync wrapper).

get_operation

def get_operation(service: str, operation: str) -> OperationSchema

Get schema for a specific operation (sync wrapper).

list_operations

def list_operations(service: str) -> list[str]

List all operations for a service (sync wrapper).

get_enum_values

def get_enum_values(service: str, operation: str,
                    element: str) -> list[str] | None

Get valid enum values for an element (sync wrapper).

list_valid_elements

def list_valid_elements(service: str, operation: str) -> list[str] | None

List all valid element names for an operation (sync wrapper).

list_cached_schemas

def list_cached_schemas() -> list[str]

List all cached service URIs.

invalidate_schema

def invalidate_schema(service: str) -> None

Invalidate a cached schema.

clear_schema_cache

def clear_schema_cache() -> None

Clear all cached schemas.

configure_ide_stubs

def configure_ide_stubs(stubs_dir: Path | str | None = None,
                        ide: str | None = None) -> str

Configure IDE to recognize xbbg type stubs.

Only modifies existing config files - does not create new ones.

Arguments:

• stubs_dir - Stubs directory (default: ~/.xbbg/stubs/)
• ide - IDE to configure: “vscode”, “pyright”, “pyproject”, “ty”, or None for auto-detect

Returns:

Message describing what was configured or manual instructions.

generate_stubs

def generate_stubs(service: str, output_dir: Path | str | None = None) -> str

Generate Python type stubs for a service.

Creates .pyi files with TypedDict definitions for request/response types. Stubs are generated locally for IDE support - never committed to repos.

Arguments:

• service - Service URI (e.g., “//blp/refdata”)
• output_dir - Output directory (default: ~/.xbbg/stubs/)

Returns:

Path to the generated stub file.