buffer - Memory buffer manager¶

Overview¶

This module provides a MemoryBuffer class for managing raw memory buffers, offering a convenient and consistent API for reading and writing various data types (integers of different sizes, strings with different termination/prefixing styles, raw bytes). It’s particularly useful for tasks involving binary data serialization/deserialization, such as implementing network protocols or handling custom file formats.

The underlying memory storage can be customized via a BufferFactory. Two factories are provided:

BytesBufferFactory: Uses Python’s built-in bytearray.
CTypesBufferFactory: Uses ctypes.create_string_buffer for potentially different memory characteristics or C-level interoperability.

Example:

from firebird.base.buffer import MemoryBuffer, ByteOrder

# Create a buffer (default uses bytearray)
buf = MemoryBuffer(10) # Initial size 10 bytes

# Write data
buf.write_short(258)       # Write 2 bytes (0x0102 in little-endian)
buf.write_pascal_string("Hi") # Write 1 byte length (2) + "Hi"
buf.write(b'\\x0A\\x0B')     # Write raw bytes

# Reset position to read
buf.pos = 0

# Read data
num = buf.read_short()
s = buf.read_pascal_string()
extra = buf.read(2)

print(f"Number: {num}")      # Output: Number: 258
print(f"String: '{s}'")      # Output: String: 'Hi'
print(f"Extra bytes: {extra}") # Output: Extra bytes: b'\\n\\x0b'
print(f"Final position: {buf.pos}") # Output: Final position: 7
print(f"Raw buffer: {buf.get_raw()}") # Output: Raw buffer: bytearray(b'\\x02\\x01\\x02Hi\\n\\x0b\\x00\\x00\\x00')

MemoryBuffer¶

class firebird.base.buffer.MemoryBuffer(init: int | bytes, size: int | None = None, *, factory: type[~firebird.base.buffer.BufferFactory] = <class 'firebird.base.buffer.BytesBufferFactory'>, eof_marker: int | None = None, max_size: int | ~firebird.base.types.Sentinel = UNLIMITED, byteorder: ~firebird.base.types.ByteOrder = ByteOrder.LITTLE)[source]¶

Bases: object

Generic memory buffer manager.

Parameters:

init (int | bytes) – Must be an integer which specifies the size of the array, or a bytes object which will be used to initialize the array items.
size (int | None) – Size of the array. The argument value is used only when init is a bytes object.
factory (type[BufferFactory]) – Factory object used to create/resize the internal memory buffer.
eof_marker (int | None) – Value that indicates the end of data. Could be None.
max_size (int | Sentinel) – If specified, the buffer couldn’t grow beyond specified number of bytes.
byteorder (ByteOrder) – The byte order used to read/write numbers.

clear() → None[source]¶

Fills the buffer with zeros and resets the position in buffer to zero.

Return type:: None

get_raw() → bytes | bytearray[source]¶

Return the underlying buffer’s content as bytes or bytearray.

Use this method for generic access to the raw buffer content instead of accessing the raw attribute directly, as the type of raw can vary depending on the buffer factory used.

Returns:: The raw content of the buffer.
Return type:: bytes | bytearray

is_eof() → bool[source]¶

Check if the current position is at or past the end of data.

End of data is defined as being beyond the buffer’s current length, or positioned exactly on a byte matching self.eof_marker (if defined).

Returns:: True if at end-of-data, False otherwise.
Return type:: bool

read(size: int = -1) → bytes[source]¶

Read specified number of bytes from current position, or all remaining data.

Advances the position by the number of bytes read.

Parameters:: size (int) – Number of bytes to read. If negative, reads all data from the current position to the end of the buffer (default: -1).
Returns:: The bytes read.
Raises:: BufferError – If size requests more bytes than available from the current position.
Return type:: bytes

read_bigint(*, signed: bool = False) → int[source]¶

Read 8 byte number (c_ulonglong).

Parameters:: signed (bool) –
Return type:: int

read_byte(*, signed: bool = False) → int[source]¶

Read 1 byte number (c_ubyte).

Parameters:: signed (bool) –
Return type:: int

read_bytes() → bytes | bytearray[source]¶

Read content of binary cluster (2 bytes data length followed by data).

Returns:: The bytes read.
Raises:: BufferError – If the end of the buffer is reached before end of data.
Return type:: bytes | bytearray

read_int(*, signed: bool = False) → int[source]¶

Read 4 byte number (c_uint).

Parameters:: signed (bool) –
Return type:: int

read_number(size: int, *, signed=False) → int[source]¶

Read a number of size bytes from current position using self.byteorder.

Advances the position by size.

Parameters:

size (int) – The number of bytes representing the number.
signed – Whether to interpret the bytes as a signed integer (default: False).

Returns:

The integer value read.

Raises:

BufferError – When size is specified, but there is not enough bytes to read.

Return type:

int

read_pascal_string(*, encoding: str = 'ascii', errors: str = 'strict') → str[source]¶

Read Pascal string (1 byte length followed by string data).

Parameters:

encoding (str) – Encoding to use for decoding (default: ‘ascii’).
errors (str) – Decoding error handling scheme (default: ‘strict’).

Returns:

The decoded string.

Raises:

BufferError – If the end of the buffer is reached before end of string.
UnicodeDecodeError – If the read bytes cannot be decoded using encoding.

Return type:

str

read_short(*, signed: bool = False) → int[source]¶

Read 2 byte number (c_ushort).

Parameters:: signed (bool) –
Return type:: int

read_sized_int(*, signed: bool = False) → int[source]¶

Read number cluster (2 byte length followed by data).

Parameters:: signed (bool) –
Return type:: int

read_sized_string(*, encoding: str = 'ascii', errors: str = 'strict') → str[source]¶

Read sized string (2 byte length followed by data).

Parameters:

encoding (str) – Encoding to use for decoding (default: ‘ascii’).
errors (str) – Decoding error handling scheme (default: ‘strict’).

Returns:

The decoded string.

Raises:

BufferError – If the end of the buffer is reached before end of string.
UnicodeDecodeError – If the read bytes cannot be decoded using encoding.

Return type:

str

read_string(*, encoding: str = 'ascii', errors: str = 'strict') → str[source]¶

Read bytes until a null terminator (0x00) is found, decode, and return string.

Advances the position past the null terminator.

Parameters:

encoding (str) – Encoding to use for decoding (default: ‘ascii’).
errors (str) – Decoding error handling scheme (default: ‘strict’).

Returns:

The decoded string (excluding the null terminator).

Raises:

BufferError – If the end of the buffer is reached before a null terminator.
UnicodeDecodeError – If the read bytes cannot be decoded using encoding.

Return type:

str

resize(size: int) → None[source]¶

Resize buffer to the specified length. Content is preserved up to the minimum of the old and new sizes. New space is uninitialized (depends on factory).

Parameters:: size (int) – The new size in bytes.
Raises:: BufferError – On attempt to resize beyond self.max_size.
Return type:: None

write(data: bytes) → None[source]¶

Write raw bytes at the current position and advance position.

Ensures buffer has enough space, resizing if necessary and allowed.

Parameters:: data (bytes) – The bytes to write.
Raises:: BufferError – If resizing is needed but exceeds max_size.
Return type:: None

write_bigint(value: int) → None[source]¶

Write 8 byte number (c_ulonglong).

Parameters:: value (int) – The integer value to write.
Raises:: BufferError – If resizing is needed but exceeds max_size.
Return type:: None

write_byte(byte: int) → None[source]¶

Write one byte.

Parameters:: byte (int) –
Return type:: None

write_int(value: int) → None[source]¶

Write 4 byte number (c_uint).

Parameters:: value (int) – The integer value to write.
Raises:: BufferError – If resizing is needed but exceeds max_size.
Return type:: None

write_number(value: int, size: int, *, signed: bool = False) → None[source]¶

Write number with specified size (in bytes).

Parameters:

value (int) – The integer value to write.
size (int) – Value size in bytes.
signed (bool) – Write as signed or unsigned integer.

Raises:

BufferError – If resizing is needed but exceeds max_size.

Return type:

None

write_pascal_string(value: str, *, encoding: str = 'ascii', errors: str = 'strict') → None[source]¶

Write Pascal string (2 byte length followed by data).

Parameters:

value (str) – The string to write.
encoding (str) – Encoding to use (default: ‘ascii’).
errors (str) – Encoding error handling scheme (default: ‘strict’).

Raises:

BufferError – If resizing is needed but exceeds max_size.

Return type:

None

write_short(value: int) → None[source]¶

Write 2 byte number (c_ushort).

Parameters:: value (int) – The integer value to write.
Raises:: BufferError – If resizing is needed but exceeds max_size.
Return type:: None

write_sized_string(value: str, *, encoding: str = 'ascii', errors: str = 'strict') → None[source]¶

Write sized string (2 byte length followed by data).

Parameters:

value (str) – The string to write.
encoding (str) – Encoding to use (default: ‘ascii’).
errors (str) – Encoding error handling scheme (default: ‘strict’).

Raises:

BufferError – If resizing is needed but exceeds max_size.

Return type:

None

write_string(value: str, *, encoding: str = 'ascii', errors: str = 'strict') → None[source]¶

Encode string, write bytes followed by a null terminator (0x00).

Parameters:

value (str) – The string to write.
encoding (str) – Encoding to use (default: ‘ascii’).
errors (str) – Encoding error handling scheme (default: ‘strict’).

Raises:

BufferError – If resizing is needed but exceeds max_size.
UnicodeEncodeError – If value cannot be encoded using encoding.

Return type:

None

property buffer_size: int¶: Current allocated buffer size in bytes.

byteorder: ByteOrder¶

LITTLE].

Type:: The byte order used to read/write numbers [default

eof_marker: int¶: Value that indicates the end of data. Could be None.

factory: BufferFactory¶

BytesBufferFactory].

Type:: Buffer factory instance used by manager [default

property last_data: int¶: Index of the last non-zero byte in the buffer (-1 if all zeros).

max_size: int | Sentinel¶

UNLIMITED].

Type:: The buffer couldn’t grow beyond specified number of bytes [default

pos: int¶: Current position in buffer, i.e. the next read/writen byte would be at this position.

raw: bytearray¶: The memory buffer. The actual data type of buffer depends on buffer factory, but it must provide direct acces to cells, slices and length like bytearray.

Buffer factories¶

class firebird.base.buffer.BufferFactory(*args, **kwargs)[source]¶

Bases: Protocol

Protocol defining the interface for creating and managing memory buffers.

Allows MemoryBuffer to work with different underlying buffer types (like bytearray or ctypes arrays).

clear(buffer: Any) → None[source]¶

Fill the buffer entirely with null bytes (zeros).

Argument:: buffer: A memory buffer previously created by this factory’s create() method.

Parameters:: buffer (Any) –
Return type:: None

create(init_or_size: int | bytes, size: int | None = None) → Any[source]¶

Create and return a mutable byte buffer object.

Parameters:

init_or_size (int | bytes) – An integer specifying the buffer size, or a bytes object for initializing the buffer content.
size (int | None) – Optional integer size, primarily used when init_or_size is bytes to specify a potentially different final size.

Returns:

The created mutable buffer object (e.g., bytearray, ctypes.c_char_Array).

Return type:

Any

get_raw(buffer: Any) → bytes | bytearray[source]¶

Return the buffer’s content as a standard bytes or bytearray.

This method is necessary to provide a consistent way to access the raw byte sequence, as the buffer object returned by create might be of a different type (e.g., ctypes arrays have a raw attribute).

Argument:: buffer: A memory buffer previously created by this factory’s create() method.

Returns:: The raw byte content of the buffer.
Parameters:: buffer (Any) –
Return type:: bytes | bytearray

class firebird.base.buffer.BytesBufferFactory[source]¶

Bases: object

Buffer factory using Python’s bytearray for storage.

clear(buffer: bytearray) → None[source]¶

Fills the bytearray buffer with zero bytes.

Parameters:: buffer (bytearray) –
Return type:: None

create(init_or_size: int | bytes, size: int | None = None) → bytearray[source]¶

This function creates a mutable character buffer. The returned object is a bytearray.

Parameters:

init_or_size (int | bytes) – Must be an integer which specifies the size of the array, or a bytes object which will be used to initialize the array items.
size (int | None) – Size of the array.

Return type:

bytearray

Important

Although arguments are the same as for ctypes.create_string_buffer, the behavior is different when new buffer is initialized from bytes:

If there are more bytes than specified size, this function copies only size bytes into new buffer. The create_string_buffer raises an excpetion.
Unlike create_string_buffer when size is NOT specified, the buffer is NOT made one item larger than its length so that the last element in the array is a NUL termination character.

get_raw(buffer: Any) → bytes | bytearray[source]¶

Returns the bytearray buffer itself.

Parameters:: buffer (Any) –
Return type:: bytes | bytearray

class firebird.base.buffer.CTypesBufferFactory[source]¶

Bases: object

Buffer factory using ctypes.create_string_buffer (array of c_char).

clear(buffer: bytearray, init: int = 0) → None[source]¶

Fills the ctypes buffer with a specified byte value using memset.

Parameters:

buffer (bytearray) – The ctypes buffer.
init (int) – The byte value to fill with (default 0).

Return type:

None

create(init_or_size: int | bytes, size: int | None = None) → bytearray[source]¶

This function creates a ctypes mutable character buffer. The returned object is an array of ctypes.c_char.

Parameters:

init_or_size (int | bytes) – Must be an integer which specifies the size of the array, or a bytes object which will be used to initialize the array items.
size (int | None) – Size of the array.

Return type:

bytearray

Important

Although arguments are the same as for ctypes.create_string_buffer, the behavior is different when new buffer is initialized from bytes:

If there are more bytes than specified size, this function copies only size bytes into new buffer. The create_string_buffer raises an excpetion.
Unlike create_string_buffer when size is NOT specified, the buffer is NOT made one item larger than its length so that the last element in the array is a NUL termination character.

get_raw(buffer: Any) → bytes | bytearray[source]¶

Returns the raw byte content via the buffer’s raw attribute.

Parameters:: buffer (Any) –
Return type:: bytes | bytearray

Functions¶

firebird.base.buffer.safe_ord(byte: bytes | int) → int[source]¶

Return the integer ordinal of a byte, or the integer itself.

Handles inputs that might already be integers (e.g., from iterating over a bytes object) or single-character bytes objects.

Parameters:: byte (bytes | int) – A single-character bytes object or an integer.
Returns:: The integer value.
Return type:: int