tessl/npm-flatbuffers
Memory efficient cross-platform serialization library with zero-copy deserialization supporting 15+ programming languages.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Securityby
Pending
The risk profile of this skill
python.mddocs/
Python
Python implementation of FlatBuffers providing a Pythonic API for serialization and deserialization while maintaining the performance benefits of zero-copy access where possible within Python's memory management model.
Capabilities
Installation and Imports
Install FlatBuffers for Python and import the necessary components.
# Installation
pip install flatbuffers
# Or with specific version
pip install flatbuffers==25.2.10# Core imports
import flatbuffers
from flatbuffers import Builder, Table
# Utility imports
from flatbuffers.compat import range_func as compat_range
from flatbuffers import encode, decodeBuilder Class
Main builder class for constructing FlatBuffer data in Python.
class Builder:
"""
FlatBuffer builder for Python
Args:
initialSize (int): Initial buffer size in bytes (default: 1024)
"""
def __init__(self, initialSize: int = 1024) -> None: ...
def Clear(self) -> None:
"""Reset the builder for reuse"""
def StartObject(self, numFields: int) -> None:
"""Start building a table/object with specified field count"""
def EndObject(self) -> int:
"""Complete table construction and return offset"""
def StartVector(self, elemSize: int, numElems: int, alignment: int) -> None:
"""Start building a vector with specified element size and count"""
def EndVector(self) -> int:
"""Complete vector construction and return offset"""
def CreateString(self, s: str) -> int:
"""Create string and return its offset"""
def CreateByteVector(self, x: bytes) -> int:
"""Create byte vector from bytes object"""
def CreateNumpyVector(self, x) -> int:
"""Create vector from numpy array"""
def PrependBool(self, x: bool) -> None:
"""Add boolean value to current vector"""
def PrependByte(self, x: int) -> None:
"""Add byte value to current vector"""
def PrependUint8(self, x: int) -> None:
"""Add unsigned 8-bit integer to current vector"""
def PrependInt8(self, x: int) -> None:
"""Add signed 8-bit integer to current vector"""
def PrependUint16(self, x: int) -> None:
"""Add unsigned 16-bit integer to current vector"""
def PrependInt16(self, x: int) -> None:
"""Add signed 16-bit integer to current vector"""
def PrependUint32(self, x: int) -> None:
"""Add unsigned 32-bit integer to current vector"""
def PrependInt32(self, x: int) -> None:
"""Add signed 32-bit integer to current vector"""
def PrependUint64(self, x: int) -> None:
"""Add unsigned 64-bit integer to current vector"""
def PrependInt64(self, x: int) -> None:
"""Add signed 64-bit integer to current vector"""
def PrependFloat32(self, x: float) -> None:
"""Add 32-bit float to current vector"""
def PrependFloat64(self, x: float) -> None:
"""Add 64-bit float to current vector"""
def PrependUOffsetTRelative(self, off: int) -> None:
"""Add offset field to current table"""
def Slot(self, slotnum: int) -> None:
"""Prepare slot for field addition"""
def Finish(self, rootTable: int, file_identifier: bytes = None) -> None:
"""Finalize buffer with root table"""
def FinishSizePrefixed(self, rootTable: int, file_identifier: bytes = None) -> None:
"""Finalize buffer with size prefix"""
def Bytes(self) -> bytes:
"""Get the finished buffer as bytes"""
def Output(self) -> bytearray:
"""Get the finished buffer as bytearray"""Usage Example:
import flatbuffers
# Create builder
builder = flatbuffers.Builder(1024)
# Create string
name = builder.CreateString("Player")
# Create vector
scores = [100, 200, 300, 400, 500]
builder.StartVector(4, len(scores), 4)
for i in reversed(range(len(scores))):
builder.PrependInt32(scores[i])
scores_offset = builder.EndVector()
# Create table
builder.StartObject(3)
builder.PrependUOffsetTRelative(name) # name field (slot 0)
builder.Slot(0)
builder.PrependInt32(42) # level field (slot 1)
builder.Slot(1)
builder.PrependUOffsetTRelative(scores_offset) # scores field (slot 2)
builder.Slot(2)
player = builder.EndObject()
# Finish buffer
builder.Finish(player)
# Get binary data
buffer = builder.Bytes()Table Class
Base class for accessing FlatBuffer table data with Python-friendly methods.
class Table:
"""Base class for FlatBuffer table access"""
def __init__(self, buf: bytes, pos: int) -> None:
"""
Initialize table accessor
Args:
buf (bytes): Buffer containing FlatBuffer data
pos (int): Position of table in buffer
"""
def Offset(self, vtableOffset: int) -> int:
"""Get field offset from vtable"""
def Get(self, flags: int, off: int) -> int:
"""Get value at offset with type flags"""
def String(self, off: int, encoding: str = 'utf-8') -> str:
"""Get string at offset"""
def VectorLen(self, off: int) -> int:
"""Get vector length at offset"""
def Vector(self, off: int) -> int:
"""Get vector starting position"""
def Union(self, off: int) -> 'Table':
"""Get union table at offset"""
def ByteVector(self, off: int) -> bytes:
"""Get byte vector at offset"""Generated Code Structure
When using flatc --python, the compiler generates Python classes following these patterns.
# Example generated Python (from monster.fbs):
import flatbuffers
from flatbuffers.compat import import_numpy
np = import_numpy()
class Vec3(object):
"""Vec3 struct definition"""
__slots__ = ['_tab']
# Struct is stored inline, so we pack directly
@staticmethod
def Pack(builder, x, y, z):
"""Pack Vec3 struct into builder"""
builder.Prep(4, 12)
builder.PrependFloat32(z)
builder.PrependFloat32(y)
builder.PrependFloat32(x)
return builder.Offset()
class Monster(object):
"""Monster table definition"""
__slots__ = ['_tab']
@classmethod
def GetRootAs(cls, buf, offset=0):
"""Get root Monster from buffer"""
n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
x = Monster()
x.Init(buf, n + offset)
return x
def Init(self, buf, pos):
"""Initialize Monster with buffer and position"""
self._tab = flatbuffers.Table(buf, pos)
def Pos(self):
"""Get position as Vec3"""
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(4))
if o != 0:
x = self._tab.Pos + o
from .Vec3 import Vec3
obj = Vec3()
obj._tab = flatbuffers.Table(self._tab.Buf, x)
return obj
return None
def Mana(self):
"""Get mana value"""
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
if o != 0:
return self._tab.Get(flatbuffers.number_types.Int16Flags, o + self._tab.Pos)
return 150 # default value
def Hp(self):
"""Get HP value"""
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(8))
if o != 0:
return self._tab.Get(flatbuffers.number_types.Int16Flags, o + self._tab.Pos)
return 100 # default value
def Name(self):
"""Get name string"""
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(10))
if o != 0:
return self._tab.String(o + self._tab.Pos)
return None
def Inventory(self, j):
"""Get inventory item at index"""
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(14))
if o != 0:
a = self._tab.Vector(o)
return self._tab.Get(flatbuffers.number_types.Uint8Flags, a + flatbuffers.number_types.UOffsetTFlags.py_type(j * 1))
return 0
def InventoryAsNumpy(self):
"""Get inventory as numpy array"""
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(14))
if o != 0:
return self._tab.GetVectorAsNumpy(flatbuffers.number_types.Uint8Flags, o)
return 0
def InventoryLength(self):
"""Get inventory vector length"""
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(14))
if o != 0:
return self._tab.VectorLen(o)
return 0
def Start(builder):
"""Start Monster construction"""
builder.StartObject(6)
def AddPos(builder, pos):
"""Add pos field to Monster"""
builder.PrependStructSlot(0, pos, 0)
def AddMana(builder, mana):
"""Add mana field to Monster"""
builder.PrependInt16Slot(1, mana, 150)
def AddHp(builder, hp):
"""Add hp field to Monster"""
builder.PrependInt16Slot(2, hp, 100)
def AddName(builder, name):
"""Add name field to Monster"""
builder.PrependUOffsetTRelativeSlot(3, name, 0)
def AddInventory(builder, inventory):
"""Add inventory field to Monster"""
builder.PrependUOffsetTRelativeSlot(5, inventory, 0)
def StartInventoryVector(builder, numElems):
"""Start inventory vector construction"""
return builder.StartVector(1, numElems, 1)
def End(builder):
"""End Monster construction"""
return builder.EndObject()NumPy Integration
FlatBuffers Python provides efficient integration with NumPy arrays for numerical computing.
import numpy as np
import flatbuffers
# Creating vectors from NumPy arrays
def create_numpy_vector(builder: flatbuffers.Builder, array: np.ndarray) -> int:
"""
Create FlatBuffer vector from NumPy array
Args:
builder: FlatBuffer builder instance
array: NumPy array to serialize
Returns:
int: Offset of created vector
"""
return builder.CreateNumpyVector(array)
# Reading vectors as NumPy arrays
def read_as_numpy(table: flatbuffers.Table, offset: int, dtype: np.dtype) -> np.ndarray:
"""
Read FlatBuffer vector as NumPy array
Args:
table: Table containing vector data
offset: Offset to vector field
dtype: NumPy data type
Returns:
np.ndarray: Array view of vector data (zero-copy when possible)
"""
return table.GetVectorAsNumpy(dtype, offset)NumPy Usage Example:
import numpy as np
import flatbuffers
# Create data
data = np.array([1.0, 2.0, 3.0, 4.0, 5.0], dtype=np.float32)
# Serialize
builder = flatbuffers.Builder(1024)
vector_offset = builder.CreateNumpyVector(data)
builder.StartObject(1)
builder.PrependUOffsetTRelativeSlot(0, vector_offset, 0)
table_offset = builder.EndObject()
builder.Finish(table_offset)
buffer = builder.Bytes()
# Deserialize
table = flatbuffers.Table(buffer, flatbuffers.encode.Get(flatbuffers.packer.uoffset, buffer, 0) + 0)
vector_data = table.GetVectorAsNumpy(flatbuffers.number_types.Float32Flags,
table.Offset(4))
print(vector_data) # [1. 2. 3. 4. 5.]Error Handling and Validation
Python FlatBuffers provides validation and error handling mechanisms.
import flatbuffers
def validate_buffer(buffer: bytes) -> bool:
"""
Validate FlatBuffer integrity
Args:
buffer (bytes): Buffer to validate
Returns:
bool: True if buffer is valid
Raises:
ValueError: If buffer is malformed
struct.error: If buffer is too short
"""
try:
# Basic validation - ensure minimum size
if len(buffer) < 4:
return False
# Check root table offset
root_offset = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buffer, 0)
if root_offset >= len(buffer):
return False
return True
except Exception:
return False
def safe_access_field(table: flatbuffers.Table, offset: int, default_value=None):
"""
Safely access table field with default fallback
Args:
table: FlatBuffer table
offset: Field offset
default_value: Value to return if field missing
Returns:
Field value or default
"""
try:
field_offset = table.Offset(offset)
if field_offset != 0:
return table.Get(flatbuffers.number_types.Int32Flags, field_offset + table.Pos)
return default_value
except (IndexError, struct.error):
return default_valueFile I/O and Serialization
Working with files and binary data in Python.
import flatbuffers
def save_flatbuffer(builder: flatbuffers.Builder, filename: str) -> None:
"""
Save FlatBuffer to file
Args:
builder: Completed FlatBuffer builder
filename: Output file path
"""
with open(filename, 'wb') as f:
f.write(builder.Bytes())
def load_flatbuffer(filename: str) -> bytes:
"""
Load FlatBuffer from file
Args:
filename: Input file path
Returns:
bytes: FlatBuffer data
"""
with open(filename, 'rb') as f:
return f.read()
def serialize_to_base64(builder: flatbuffers.Builder) -> str:
"""
Convert FlatBuffer to base64 string
Args:
builder: Completed FlatBuffer builder
Returns:
str: Base64 encoded string
"""
import base64
return base64.b64encode(builder.Bytes()).decode('ascii')
def deserialize_from_base64(data: str) -> bytes:
"""
Convert base64 string to FlatBuffer
Args:
data: Base64 encoded string
Returns:
bytes: FlatBuffer data
"""
import base64
return base64.b64decode(data.encode('ascii'))Complete Usage Example:
import flatbuffers
from generated.Monster import Monster, Start, End, AddName, AddHp, AddMana, AddPos
from generated.Vec3 import Vec3
# Create monster
builder = flatbuffers.Builder(1024)
# Create components
name_offset = builder.CreateString("Dragon")
pos_offset = Vec3.Pack(builder, 1.0, 2.0, 3.0)
# Build monster table
Start(builder)
AddName(builder, name_offset)
AddHp(builder, 200)
AddMana(builder, 150)
AddPos(builder, pos_offset)
monster_offset = End(builder)
# Finish buffer
builder.Finish(monster_offset)
buffer = builder.Bytes()
# Read the data back
monster = Monster.GetRootAs(buffer, 0)
print(f"Name: {monster.Name()}")
print(f"HP: {monster.Hp()}")
print(f"Mana: {monster.Mana()}")
pos = monster.Pos()
if pos:
print(f"Position: {pos.X()}, {pos.Y()}, {pos.Z()}")
# Save to file
with open('monster.bin', 'wb') as f:
f.write(buffer)
# Load from file
with open('monster.bin', 'rb') as f:
loaded_buffer = f.read()
loaded_monster = Monster.GetRootAs(loaded_buffer, 0)
print(f"Loaded name: {loaded_monster.Name()}")