Module Contents¶

Classes¶

`FbsType`	Flatbuffers type.
`FbsAttribute`
`FbsField`
`FbsObject`
`FbsRPCCall`
`FbsService`
`FbsEnumValue`
`FbsEnum`	FlatBuffers enum type.
`FbsSchema`
`FbsRepository`	crossbar.interfaces.IInventory

Functions¶

`parse_attr`(obj)
`parse_docs`(obj)
`parse_fields`(repository, schema, obj[, objs_lst])
`parse_calls`(repository, schema, svc_obj[, objs_lst])
`validate_scalar`(field, value)

class autobahn.xbr._schema.FbsType(repository: FbsRepository, schema: FbsSchema, basetype: int, element: int, index: int, objtype: Optional[str] = None, elementtype: Optional[str] = None)[source]¶

Bases: object

Flatbuffers type.

See: https://github.com/google/flatbuffers/blob/11a19887053534c43f73e74786b46a615ecbf28e/reflection/reflection.fbs#L33

property repository: FbsRepository¶

property schema: FbsSchema¶

property basetype: int¶

Flatbuffers base type.

Returns

property element: int¶

Only if basetype == Vector

Returns

property index: int¶

If basetype == Object, index into “objects”. If base_type == Union, UnionType, or integral derived from an enum, index into “enums”. If base_type == Vector && element == Union or UnionType.

Returns

property elementtype: Optional[str]¶

If basetype == Vector, fully qualified element type name.

Returns

property objtype: Optional[str]¶

If basetype == Object, fully qualified object type name.

Returns

__slots__ = ['_repository', '_schema', '_basetype', '_element', '_index', '_objtype', '_elementtype']¶

UType¶

Bool¶

Byte¶

UByte¶

Short¶

UShort¶

Int¶

UInt¶

Long¶

ULong¶

Float¶

Double¶

String¶

SCALAR_TYPES¶

Vector¶

Obj¶

Union¶

STRUCTURED_TYPES¶

FBS2PY¶

FBS2PY_TYPE¶

FBS2FLAGS¶

FBS2PREPEND¶

FBS2STR¶

STR2FBS¶

map(language: str, attrs: Optional[Dict] = None, required: Optional[bool] = True, objtype_as_string: bool = False) → str[source]¶

Parameters

language –
attrs –
required –
objtype_as_string –

Returns

__str__() → str[source]¶: Return str(self).

marshal() → Dict[str, Any][source]¶

class autobahn.xbr._schema.FbsAttribute[source]¶

Bases: object

__str__()[source]¶: Return str(self).

class autobahn.xbr._schema.FbsField(repository: FbsRepository, schema: FbsSchema, name: str, type: FbsType, id: int, offset: int, default_int: int, default_real: float, deprecated: bool, required: bool, attrs: Dict[str, FbsAttribute], docs: str)[source]¶

Bases: object

property repository: FbsRepository¶

property schema: FbsSchema¶

property name: str¶

property type: FbsType¶

property id: int¶

property offset: int¶

property default_int: int¶

property default_real: float¶

property deprecated: bool¶

property required: bool¶

property attrs: Dict[str, FbsAttribute]¶

property docs: str¶

__slots__ = ['_repository', '_schema', '_name', '_type', '_id', '_offset', '_default_int', '_default_real',...¶

__str__() → str[source]¶: Return str(self).

marshal() → Dict[str, Any][source]¶

autobahn.xbr._schema.parse_attr(obj)[source]¶

autobahn.xbr._schema.parse_docs(obj)[source]¶

autobahn.xbr._schema.parse_fields(repository, schema, obj, objs_lst=None)[source]¶

autobahn.xbr._schema.parse_calls(repository, schema, svc_obj, objs_lst=None)[source]¶

class autobahn.xbr._schema.FbsObject(repository: FbsRepository, schema: FbsSchema, declaration_file: str, name: str, fields: Dict[str, FbsField], fields_by_id: List[FbsField], is_struct: bool, min_align: int, bytesize: int, attrs: Dict[str, FbsAttribute], docs: str)[source]¶

Bases: object

property repository: FbsRepository¶

property schema: FbsSchema¶

property declaration_file: str¶

property name: str¶

property fields: Dict[str, FbsField]¶

property fields_by_id: List[FbsField]¶

property is_struct: bool¶

property min_align: int¶

property bytesize: int¶

property attrs: Dict[str, FbsAttribute]¶

property docs: str¶

__slots__ = ['_repository', '_schema', '_declaration_file', '_name', '_fields', '_fields_by_id',...¶

map(language: str, required: Optional[bool] = True, objtype_as_string: bool = False) → str[source]¶

map_import(language: str) → str[source]¶

__str__() → str[source]¶: Return str(self).

marshal() → Dict[str, Any][source]¶

static parse(repository, schema, fbs_obj, objs_lst=None)[source]¶

class autobahn.xbr._schema.FbsRPCCall(repository: FbsRepository, schema: FbsSchema, name: str, id: int, request: FbsObject, response: FbsObject, docs: str, attrs: Dict[str, FbsAttribute])[source]¶

Bases: object

property repository¶

property schema¶

property name¶

property id¶

property request¶

property response¶

property docs¶

property attrs¶

__str__()[source]¶: Return str(self).

marshal()[source]¶

class autobahn.xbr._schema.FbsService(repository: FbsRepository, schema: FbsSchema, declaration_file: str, name: str, calls: Dict[str, FbsRPCCall], calls_by_id: List[FbsRPCCall], attrs: Dict[str, FbsAttribute], docs: str)[source]¶

Bases: object

property repository¶

property schema¶

property declaration_file¶

property name¶

property calls¶

property calls_by_id¶

property attrs¶

property docs¶

__str__()[source]¶: Return str(self).

marshal()[source]¶

class autobahn.xbr._schema.FbsEnumValue(repository: FbsRepository, schema: FbsSchema, name: str, id: int, value, docs)[source]¶

Bases: object

property repository¶

property schema¶

property name¶

property id¶

property value¶

property attrs¶

property docs¶

__str__()[source]¶: Return str(self).

marshal()[source]¶

class autobahn.xbr._schema.FbsEnum(repository: FbsRepository, schema: FbsSchema, declaration_file: str, name: str, id: int, values: Dict[str, FbsEnumValue], values_by_id: List[FbsEnumValue], is_union: bool, underlying_type: int, attrs: Dict[str, FbsAttribute], docs: str)[source]¶

Bases: object

FlatBuffers enum type.

See https://github.com/google/flatbuffers/blob/11a19887053534c43f73e74786b46a615ecbf28e/reflection/reflection.fbs#L61

property repository¶

property schema¶

property declaration_file¶

property name¶

property id¶

property values¶

property values_by_id¶

property is_union¶

property underlying_type¶

property attrs¶

property docs¶

__str__()[source]¶: Return str(self).

marshal()[source]¶

class autobahn.xbr._schema.FbsSchema(repository: FbsRepository, file_name: str, file_sha256: str, file_size: int, file_ident: str, file_ext: str, fbs_files: List[Dict[str, str]], root_table: FbsObject, root: zlmdb.flatbuffers.reflection.Schema.Schema, objs: Optional[Dict[str, FbsObject]] = None, objs_by_id: Optional[List[FbsObject]] = None, enums: Optional[Dict[str, FbsEnum]] = None, enums_by_id: Optional[List[FbsEnum]] = None, services: Optional[Dict[str, FbsService]] = None, services_by_id: Optional[List[FbsService]] = None)[source]¶

Bases: object

property repository¶

property file_name¶

property file_sha256¶

property file_size¶

property file_ident¶

property file_ext¶

property fbs_files¶

property root_table¶

property root¶

property objs¶

property objs_by_id¶

property enums¶

property enums_by_id¶

property services¶

property services_by_id¶

__str__()[source]¶: Return str(self).

marshal() → Dict[str, object][source]¶

Returns

static load(repository: FbsRepository, sfile: Union[str, io.RawIOBase, IO[bytes]], filename: Optional[str] = None) → FbsSchema[source]¶

Parameters

repository –
sfile –
filename –

Returns

autobahn.xbr._schema.validate_scalar(field, value: Optional[Any])[source]¶

class autobahn.xbr._schema.FbsRepository(basemodule: str)[source]¶

Bases: object

crossbar.interfaces.IInventory

add: FbsRepository[] - load: FbsSchema[]

https://github.com/google/flatbuffers/blob/master/reflection/reflection.fbs

property basemodule: str¶

property schemas: Dict[str, FbsSchema]¶

property objs: Dict[str, FbsObject]¶

property enums: Dict[str, FbsEnum]¶

property services: Dict[str, FbsService]¶

property total_count¶

static from_archive(filename: str) → FbsRepository[source]¶

static from_address(address: str) → FbsRepository[source]¶

load(filename: str) → Tuple[int, int][source]¶

Load and add all schemata from Flatbuffers binary schema files (*.bfbs) found in the given directory. Alternatively, a path to a single schema file can be provided.

Parameters: filename – Filesystem path of a directory or single file from which to load and add Flatbuffers schemata.

summary(keys=False)[source]¶

print_summary()[source]¶

render(jinja2_env, output_dir, output_lang)[source]¶

Parameters

jinja2_env –
output_dir –
output_lang –

Returns

validate_obj(validation_type: Optional[str], value: Optional[Any])[source]¶

Validate value against the validation type given.

If the application payload does not validate against the provided type, an autobahn.wamp.exception.InvalidPayload is raised.

Parameters

validation_type – Flatbuffers type (fully qualified) against to validate application payload.
value – Value to validate.

Returns

validate(validation_type: str, args: List[Any], kwargs: Dict[str, Any]) → Optional[FbsObject][source]¶

Validate the WAMP application payload provided in positional argument in args and in keyword-based arguments in kwargs against the FlatBuffers table type validation_type from this repository.

If the application payload does not validate against the provided type, an autobahn.wamp.exception.InvalidPayload is raised.

Parameters

validation_type – Flatbuffers type (fully qualified) against to validate application payload.
args – The application payload WAMP positional arguments.
kwargs – The application payload WAMP keyword-based arguments.

Returns

The validation type object from this repository (reference in validation_type) which has been used for validation.