Skip to content

πŸ”΄ unaiverse.networking.p2p.lib_types

What this module does πŸ”΄

Provides the TypeInterface marshalling layer that converts Python values to and from Go ctypes representations (strings, ints, floats, bools, bytes, JSON) and manages freeing of Go-allocated pointers.

lib_types

β–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆβ–‘β–‘β–‘β–‘β–‘β–ˆβ–‘β–‘β–ˆβ–ˆβ–ˆβ–‘β–‘β–‘β–‘β–‘β–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–‘β–‘β–‘β–‘β–‘β–ˆβ–ˆβ–ˆβ–‘β–‘β–ˆβ–ˆβ–ˆβ–‘β–‘β–‘β–‘β–‘β–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆβ–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–ˆ β–‘ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–‘ β–‘β–ˆβ–ˆβ–ˆ β–ˆ β–‘ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆβ–‘β–‘β–ˆβ–ˆβ–ˆβ–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–‘β–‘β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ
β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆβ–‘β–‘β–ˆ β–‘β–ˆβ–ˆβ–ˆβ–‘β–‘β–‘β–‘β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆβ–‘β–‘β–ˆ
β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–‘β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–‘β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–‘ β–‘β–ˆβ–ˆβ–ˆ β–‘ β–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘β–ˆβ–ˆβ–ˆ β–‘ β–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–‘β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–‘β–‘β–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–‘β–‘β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘ β–‘β–‘β–‘β–‘β–‘ β–‘β–‘β–‘β–‘β–‘ β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘ β–‘β–‘β–‘β–‘β–‘ β–‘β–‘β–‘ β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘ β–‘β–‘β–‘β–‘β–‘ β–‘β–‘β–‘β–‘β–‘ β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘ β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘β–‘ A Collectionless AI Project (https://collectionless.ai) Registration/Login: https://unaiverse.io Code Repositories: https://github.com/collectionlessai/ Main Developers: Stefano Melacci (Project Leader), Christian Di Maio, Tommaso Guidi

logger module-attribute

logger = getLogger('LIB-TYPES')

TypeInterface

TypeInterface(libp2p_instance: GoLibP2P)

Low-level bridge for converting between Python types and Go/C types via ctypes.

TypeInterface centralises all marshalling and unmarshalling needed when calling functions exported by a Go shared library through ctypes. Every method either converts a Python value into a C-compatible representation ready to pass into a Go function, or converts a value returned by Go back into a native Python type.

A particular concern is memory safety: Go functions that return heap-allocated C strings transfer ownership of that memory to the caller. from_go_ptr_to_json therefore calls FreeString on the underlying GoLibP2P instance to release the memory, and maintains an internal freed-pointer registry protected by a threading.Lock to guard against accidental double-free errors.

Attributes:

Name Type Description
libp2p GoLibP2P

The GoLibP2P shared-library wrapper used to call Go-exported functions, including FreeString for releasing C memory.

Initialize the type bridge with a loaded Go shared-library instance.

An internal freed-pointer registry (a set guarded by a threading.Lock) is created to detect and prevent double-free errors across concurrent calls to from_go_ptr_to_json.

Parameters:

Name Type Description Default
libp2p_instance GoLibP2P

A fully loaded GoLibP2P instance whose exported functions (including FreeString) will be used for all marshalling operations.

required
Source code in unaiverse/networking/p2p/lib_types.py
def __init__(self, libp2p_instance: GoLibP2P):
    """Initialize the type bridge with a loaded Go shared-library instance.

    An internal freed-pointer registry (a ``set`` guarded by a ``threading.Lock``)
    is created to detect and prevent double-free errors across concurrent calls
    to ``from_go_ptr_to_json``.

    Args:
        libp2p_instance: A fully loaded ``GoLibP2P`` instance whose exported
            functions (including ``FreeString``) will be used for all
            marshalling operations.
    """
    self.__freed_pointers: set[int] = set()  # Track freed pointers to prevent double-free errors
    self.__freed_pointers_lock: Any = Lock()  # A threading lock
    self.libp2p: GoLibP2P = libp2p_instance  # Store the shared library object instance

libp2p instance-attribute

libp2p: GoLibP2P = libp2p_instance

to_go_string

to_go_string(s: str) -> bytes

Convert a Python string to a UTF-8 encoded bytes object for Go/C interop.

The returned bytes value is suitable for direct use with ctypes when passing to a C function that expects a char* (ctypes.c_char_p): ctypes automatically passes a pointer to the byte string's internal data buffer.

If s is None it is treated as an empty string, so callers do not need to guard against None before invoking this method.

Parameters:

Name Type Description Default
s str

The Python string to encode. None is treated as an empty string.

required

Returns:

Type Description
bytes

A bytes object containing the UTF-8 encoded representation of s.

Source code in unaiverse/networking/p2p/lib_types.py
def to_go_string(self, s: str) -> bytes:
    """Convert a Python string to a UTF-8 encoded ``bytes`` object for Go/C interop.

    The returned ``bytes`` value is suitable for direct use with ctypes when passing
    to a C function that expects a ``char*`` (``ctypes.c_char_p``): ctypes
    automatically passes a pointer to the byte string's internal data buffer.

    If ``s`` is ``None`` it is treated as an empty string, so callers do not need
    to guard against ``None`` before invoking this method.

    Args:
        s: The Python string to encode. ``None`` is treated as an empty string.

    Returns:
        A ``bytes`` object containing the UTF-8 encoded representation of ``s``.
    """
    if s is None:
        s = ""
    return s.encode("utf-8")

from_go_string

from_go_string(cstr: bytes) -> str

Convert a null-terminated C string returned by Go into a Python str.

The argument is expected to be the raw value obtained from a ctypes.c_char_p result type, which ctypes exposes as a bytes object (or None for a NULL pointer). A falsy value (None or empty bytes) is mapped to an empty string.

Parameters:

Name Type Description Default
cstr bytes

The raw bytes value from a ctypes.c_char_p result, or None for a NULL pointer.

required

Returns:

Type Description
str

The UTF-8 decoded Python string, or an empty string when cstr is falsy.

Source code in unaiverse/networking/p2p/lib_types.py
def from_go_string(self, cstr: bytes) -> str:
    """Convert a null-terminated C string returned by Go into a Python ``str``.

    The argument is expected to be the raw value obtained from a ``ctypes.c_char_p``
    result type, which ctypes exposes as a ``bytes`` object (or ``None`` for a NULL
    pointer). A falsy value (``None`` or empty bytes) is mapped to an empty string.

    Args:
        cstr: The raw bytes value from a ``ctypes.c_char_p`` result, or ``None``
            for a NULL pointer.

    Returns:
        The UTF-8 decoded Python string, or an empty string when ``cstr`` is falsy.
    """
    if not cstr:
        return ""
    return cstr.decode("utf-8")

to_go_int

to_go_int(i: int) -> c_int

Convert a Python integer to a Go-compatible ctypes.c_int.

Parameters:

Name Type Description Default
i int

The Python integer to convert.

required

Returns:

Type Description
c_int

A ctypes.c_int wrapping i, suitable for passing to a Go-exported

c_int

function that expects a C int.

Source code in unaiverse/networking/p2p/lib_types.py
def to_go_int(self, i: int) -> ctypes.c_int:
    """Convert a Python integer to a Go-compatible ``ctypes.c_int``.

    Args:
        i: The Python integer to convert.

    Returns:
        A ``ctypes.c_int`` wrapping ``i``, suitable for passing to a Go-exported
        function that expects a C ``int``.
    """
    return ctypes.c_int(i)

from_go_int

from_go_int(val: c_int) -> int

Convert a ctypes.c_int returned by Go into a Python int.

Parameters:

Name Type Description Default
val c_int

The ctypes.c_int value returned from a Go-exported function.

required

Returns:

Type Description
int

The corresponding Python int.

Source code in unaiverse/networking/p2p/lib_types.py
def from_go_int(self, val: ctypes.c_int) -> int:
    """Convert a ``ctypes.c_int`` returned by Go into a Python ``int``.

    Args:
        val: The ``ctypes.c_int`` value returned from a Go-exported function.

    Returns:
        The corresponding Python ``int``.
    """
    return int(val)

to_go_float

to_go_float(f: float) -> c_float

Convert a Python float to a Go-compatible ctypes.c_float.

Parameters:

Name Type Description Default
f float

The Python float to convert.

required

Returns:

Type Description
c_float

A ctypes.c_float wrapping f, suitable for passing to a Go-exported

c_float

function that expects a C float.

Source code in unaiverse/networking/p2p/lib_types.py
def to_go_float(self, f: float) -> ctypes.c_float:
    """Convert a Python float to a Go-compatible ``ctypes.c_float``.

    Args:
        f: The Python float to convert.

    Returns:
        A ``ctypes.c_float`` wrapping ``f``, suitable for passing to a Go-exported
        function that expects a C ``float``.
    """
    return ctypes.c_float(f)

from_go_float

from_go_float(val: c_float) -> float

Convert a ctypes.c_float returned by Go into a Python float.

Parameters:

Name Type Description Default
val c_float

The ctypes.c_float value returned from a Go-exported function.

required

Returns:

Type Description
float

The corresponding Python float.

Source code in unaiverse/networking/p2p/lib_types.py
def from_go_float(self, val: ctypes.c_float) -> float:
    """Convert a ``ctypes.c_float`` returned by Go into a Python ``float``.

    Args:
        val: The ``ctypes.c_float`` value returned from a Go-exported function.

    Returns:
        The corresponding Python ``float``.
    """
    return float(val)

to_go_bool

to_go_bool(b: bool) -> c_int

Convert a Python boolean to a Go-compatible integer (1 for True, 0 for False).

Go does not export a dedicated bool type across the C boundary; booleans are conventionally represented as int values where non-zero means true.

Parameters:

Name Type Description Default
b bool

The Python boolean to convert.

required

Returns:

Type Description
c_int

A ctypes.c_int with value 1 when b is True, or 0

c_int

when b is False.

Source code in unaiverse/networking/p2p/lib_types.py
def to_go_bool(self, b: bool) -> ctypes.c_int:
    """Convert a Python boolean to a Go-compatible integer (``1`` for True, ``0`` for False).

    Go does not export a dedicated bool type across the C boundary; booleans are
    conventionally represented as ``int`` values where non-zero means true.

    Args:
        b: The Python boolean to convert.

    Returns:
        A ``ctypes.c_int`` with value ``1`` when ``b`` is ``True``, or ``0``
        when ``b`` is ``False``.
    """
    return ctypes.c_int(1 if b else 0)

from_go_bool

from_go_bool(val: c_int) -> bool

Convert a Go-compatible integer (ctypes.c_int) to a Python boolean.

The convention is that a value of exactly 1 means True; any other value (including negative integers) is treated as False.

Parameters:

Name Type Description Default
val c_int

The ctypes.c_int value returned from a Go-exported function.

required

Returns:

Type Description
bool

True if val equals 1, False otherwise.

Source code in unaiverse/networking/p2p/lib_types.py
def from_go_bool(self, val: ctypes.c_int) -> bool:
    """Convert a Go-compatible integer (``ctypes.c_int``) to a Python boolean.

    The convention is that a value of exactly ``1`` means ``True``; any other
    value (including negative integers) is treated as ``False``.

    Args:
        val: The ``ctypes.c_int`` value returned from a Go-exported function.

    Returns:
        ``True`` if ``val`` equals ``1``, ``False`` otherwise.
    """
    return val == 1

to_go_bytes

to_go_bytes(b: bytes) -> c_char_p

Convert a Python bytes object to a Go-compatible ctypes.c_char_p.

A fixed-size C buffer is allocated via ctypes.create_string_buffer and then cast to ctypes.c_char_p so that Go receives a stable pointer to the byte data. If b is None, an empty byte string is used instead.

Parameters:

Name Type Description Default
b bytes

The Python bytes to convert. None is treated as empty bytes.

required

Returns:

Type Description
c_char_p

A ctypes.c_char_p pointing to the contents of b.

Source code in unaiverse/networking/p2p/lib_types.py
def to_go_bytes(self, b: bytes) -> ctypes.c_char_p:
    """Convert a Python ``bytes`` object to a Go-compatible ``ctypes.c_char_p``.

    A fixed-size C buffer is allocated via ``ctypes.create_string_buffer`` and
    then cast to ``ctypes.c_char_p`` so that Go receives a stable pointer to the
    byte data. If ``b`` is ``None``, an empty byte string is used instead.

    Args:
        b: The Python ``bytes`` to convert. ``None`` is treated as empty bytes.

    Returns:
        A ``ctypes.c_char_p`` pointing to the contents of ``b``.
    """
    if b is None:
        b = b""
    buf = ctypes.create_string_buffer(b, len(b))
    return ctypes.cast(buf, ctypes.c_char_p)

from_go_bytes

from_go_bytes(cptr: c_char_p, length: int) -> bytes

Convert a raw C byte buffer returned by Go into a Python bytes object.

ctypes.string_at is used to copy exactly length bytes starting at the address given by cptr. If either cptr is a NULL pointer or length is not positive, an empty bytes object is returned immediately without attempting a memory read.

Parameters:

Name Type Description Default
cptr c_char_p

The ctypes.c_char_p (or integer address) pointing to the byte buffer allocated by Go.

required
length int

The number of bytes to read from cptr.

required

Returns:

Type Description
bytes

A bytes object containing the copied data, or an empty bytes

bytes

when cptr is falsy or length is not positive.

Source code in unaiverse/networking/p2p/lib_types.py
def from_go_bytes(self, cptr: ctypes.c_char_p, length: int) -> bytes:
    """Convert a raw C byte buffer returned by Go into a Python ``bytes`` object.

    ``ctypes.string_at`` is used to copy exactly ``length`` bytes starting at
    the address given by ``cptr``. If either ``cptr`` is a NULL pointer or
    ``length`` is not positive, an empty ``bytes`` object is returned immediately
    without attempting a memory read.

    Args:
        cptr: The ``ctypes.c_char_p`` (or integer address) pointing to the byte
            buffer allocated by Go.
        length: The number of bytes to read from ``cptr``.

    Returns:
        A ``bytes`` object containing the copied data, or an empty ``bytes``
        when ``cptr`` is falsy or ``length`` is not positive.
    """
    if not cptr or length <= 0:
        return bytes()
    return ctypes.string_at(cptr, length)

from_go_ptr_to_json

from_go_ptr_to_json(c_void_ptr_val: int) -> Any

Convert a Go-allocated C string containing JSON into a Python object, then free the memory.

The full lifecycle is: validate the pointer is not NULL and has not already been freed; cast the integer address to ctypes.c_char_p and read the null-terminated byte string; decode it as UTF-8; parse the resulting string as JSON; and - in a finally block that executes even on error - call libp2p.FreeString to release the C heap memory allocated by Go.

A freed-pointer registry (protected by an internal threading.Lock) guards against double-free errors. If the same pointer address is passed a second time before the registry is cleared, the method raises instead of calling FreeString again.

Parameters:

Name Type Description Default
c_void_ptr_val int

The integer address of the C string allocated by Go. A value of 0 is interpreted as a NULL pointer and raises immediately.

required

Returns:

Type Description
Any

The Python object produced by parsing the JSON string (typically a

Any

dict or a list).

Raises:

Type Description
Exception

If c_void_ptr_val is 0 (NULL pointer).

Exception

If the pointer was already freed (detected via the internal registry), indicating a double-free logic error.

Exception

If reading or UTF-8 decoding the C string fails.

Exception

If the string is not valid JSON (json.JSONDecodeError is re-raised as a plain Exception).

Note

FreeString is always called in the finally block regardless of whether JSON parsing succeeds or fails, to avoid memory leaks. If FreeString itself raises, the error is logged at CRITICAL level but is not re-raised so that the original exception (if any) propagates unchanged.

Source code in unaiverse/networking/p2p/lib_types.py
def from_go_ptr_to_json(self, c_void_ptr_val: int) -> Any:
    """Convert a Go-allocated C string containing JSON into a Python object, then free the memory.

    The full lifecycle is: validate the pointer is not NULL and has not already
    been freed; cast the integer address to ``ctypes.c_char_p`` and read the
    null-terminated byte string; decode it as UTF-8; parse the resulting string
    as JSON; and - in a ``finally`` block that executes even on error - call
    ``libp2p.FreeString`` to release the C heap memory allocated by Go.

    A freed-pointer registry (protected by an internal ``threading.Lock``) guards
    against double-free errors. If the same pointer address is passed a second
    time before the registry is cleared, the method raises instead of calling
    ``FreeString`` again.

    Args:
        c_void_ptr_val: The integer address of the C string allocated by Go.
            A value of ``0`` is interpreted as a NULL pointer and raises
            immediately.

    Returns:
        The Python object produced by parsing the JSON string (typically a
        ``dict`` or a ``list``).

    Raises:
        Exception: If ``c_void_ptr_val`` is ``0`` (NULL pointer).
        Exception: If the pointer was already freed (detected via the internal
            registry), indicating a double-free logic error.
        Exception: If reading or UTF-8 decoding the C string fails.
        Exception: If the string is not valid JSON (``json.JSONDecodeError``
            is re-raised as a plain ``Exception``).

    Note:
        ``FreeString`` is always called in the ``finally`` block regardless of
        whether JSON parsing succeeds or fails, to avoid memory leaks. If
        ``FreeString`` itself raises, the error is logged at ``CRITICAL`` level
        but is not re-raised so that the original exception (if any) propagates
        unchanged.
    """

    if not c_void_ptr_val:  # Check if the address is NULL (0)
        print("Received a NULL pointer from Go function")
        raise Exception("Received a NULL pointer from Go function")

    try:

        # --- Double-Free Check (Before Reading/Casting) ---
        self.__freed_pointers_lock.acquire()  # Acquire lock if using threading
        if c_void_ptr_val in self.__freed_pointers:

            # This indicates a serious logic error elsewhere - the pointer
            # was already freed but somehow passed here again.
            logger.warning(f"πŸ”₯πŸ”₯πŸ”₯ ATTEMPT TO PROCESS ALREADY FREED POINTER {hex(c_void_ptr_val)}! πŸ”₯πŸ”₯πŸ”₯")

            # Raising an error is safer than trying to read potentially invalid memory.
            logger.error(f"Attempt to process pointer {hex(c_void_ptr_val)} which was already freed")
            raise Exception(f"Attempt to process pointer {hex(c_void_ptr_val)} which was already freed")
        self.__freed_pointers_lock.release()  # Release lock if using threading

        # --- Cast void* to c_char_p and Read String ---
        try:

            # Perform the cast only when needed for reading
            c_char_ptr_for_read = ctypes.cast(c_void_ptr_val, ctypes.c_char_p)
            raw_bytes = ctypes.string_at(c_char_ptr_for_read)
            json_string = raw_bytes.decode('utf-8')

            # Logger.debug(f"Read string (len={len(json_string)})
            # from pointer {hex(c_void_ptr_val)}: %.100s...", json_string)
        except (ctypes.ArgumentError, ValueError, UnicodeDecodeError) as read_err:
            logger.error(f"Failed to read/decode string from pointer {hex(c_void_ptr_val)}: "
                         f"{read_err}", exc_info=False)

            # Even if reading fails, the pointer itself *might* still be valid C memory
            # that Go expects us to free. We will proceed to free it in finally.
            raise Exception(f"Failed to read string from pointer {hex(c_void_ptr_val)}: {read_err}") from read_err
        except Exception as unexpected_read_err:  # Catch other potential ctypes issues
            logger.error(f"Unexpected error reading C string from pointer {hex(c_void_ptr_val)}: "
                         f"{unexpected_read_err}", exc_info=True)
            raise Exception(f"Unexpected error reading C string from pointer {hex(c_void_ptr_val)}: "
                            f"{unexpected_read_err}") from unexpected_read_err

        # --- Check for Empty String ---

        # --- Parse JSON ---
        try:

            # Now that we have the string, parse it
            logger.debug(f"Parsing JSON from string: {json_string}")
            parsed_data = json.loads(json_string)
            logger.debug(f"Parsed JSON data: {parsed_data}")

            # Logger.debug(f"Successfully parsed JSON from pointer {hex(c_void_ptr_val)}")
            return parsed_data  # Return the parsed Python object

        except json.JSONDecodeError as json_err:
            logger.error(f"Failed to decode JSON from pointer {hex(c_void_ptr_val)}: {json_err}", exc_info=False)

            # Again, the pointer is likely valid C memory, but the content is bad.
            # Let the block handle freeing.
            raise Exception(f"Failed to decode JSON from pointer {hex(c_void_ptr_val)}: {json_err}") from json_err

    finally:

        # --- CRITICAL: Free C Memory ---
        # This block executes even if errors occurred during read/parse,
        # ensuring we attempt to free any non-NULL pointer received from Go.
        with self.__freed_pointers_lock:
            if c_void_ptr_val:
                logger.info(f"🐍 FINALLY: Freeing pointer {hex(c_void_ptr_val)}...")
                if c_void_ptr_val in self.__freed_pointers:

                    # This check is technically redundant if the initial check worked,
                    # but provides an extra safety layer in case of concurrency issues
                    # (if freed_pointers is shared without locks - which it shouldn't be).
                    logger.warning(f"πŸ”₯πŸ”₯πŸ”₯ DOUBLE FREE DETECTED in finally block for "
                                   f"{hex(c_void_ptr_val)}! Skipping FreeString call again. πŸ”₯πŸ”₯πŸ”₯")
                else:

                    # Add before calling free
                    try:
                        self.libp2p.FreeString(c_void_ptr_val)  # Pass the original void* value
                        # self.__freed_pointers_lock.add(c_void_ptr_val)  # TODO do we still need this?
                        logger.info(f"βœ… FINALLY: FreeString successful for {hex(c_void_ptr_val)}.")
                    except Exception as free_err:

                        # Log if FreeString fails, but don't raise from finally
                        # as it might hide the original error.
                        logger.critical(f"🚨 FAILED TO FREE C MEMORY for pointer "
                                        f"{hex(c_void_ptr_val)} via FreeString: {free_err}", exc_info=True)

to_go_json

to_go_json(data: Any) -> bytes

Encode a Python object to a JSON string and return it as UTF-8 bytes for Go/C interop.

The object is serialised with json.dumps and the resulting string is then passed through to_go_string to produce a UTF-8 encoded bytes value. The returned value is suitable for direct use with ctypes when passing to a C function that expects a char* (ctypes.c_char_p).

Parameters:

Name Type Description Default
data Any

The Python object to serialise (for example a dict or list).

required

Returns:

Type Description
bytes

A bytes object containing the UTF-8 encoded JSON representation of

bytes

data, ready to be passed to a Go-exported C function.

Source code in unaiverse/networking/p2p/lib_types.py
def to_go_json(self, data: Any) -> bytes:
    """Encode a Python object to a JSON string and return it as UTF-8 ``bytes`` for Go/C interop.

    The object is serialised with ``json.dumps`` and the resulting string is then
    passed through ``to_go_string`` to produce a UTF-8 encoded ``bytes`` value.
    The returned value is suitable for direct use with ctypes when passing to a C
    function that expects a ``char*`` (``ctypes.c_char_p``).

    Args:
        data: The Python object to serialise (for example a ``dict`` or ``list``).

    Returns:
        A ``bytes`` object containing the UTF-8 encoded JSON representation of
        ``data``, ready to be passed to a Go-exported C function.
    """
    json_str = json.dumps(data)
    return self.to_go_string(json_str)

from_go_string_to_list

from_go_string_to_list(cstr: c_char_p) -> List[Any]

Decode a JSON-encoded list from a Go C string into a Python list.

The C string is first converted to a Python str via from_go_string, and the resulting JSON text is then parsed with json.loads. The Go-exported function is expected to produce a JSON array; if the content is not a valid JSON list the call will raise.

Parameters:

Name Type Description Default
cstr c_char_p

The raw ctypes.c_char_p value (or equivalent bytes) returned by a Go-exported function, containing a UTF-8 encoded JSON array.

required

Returns:

Type Description
List[Any]

A Python list parsed from the JSON content of cstr.

Raises:

Type Description
JSONDecodeError

If cstr does not contain valid JSON.

UnicodeDecodeError

If the byte sequence in cstr is not valid UTF-8.

Source code in unaiverse/networking/p2p/lib_types.py
def from_go_string_to_list(self, cstr: ctypes.c_char_p) -> List[Any]:
    """Decode a JSON-encoded list from a Go C string into a Python ``list``.

    The C string is first converted to a Python ``str`` via ``from_go_string``, and
    the resulting JSON text is then parsed with ``json.loads``. The Go-exported
    function is expected to produce a JSON array; if the content is not a valid JSON
    list the call will raise.

    Args:
        cstr: The raw ``ctypes.c_char_p`` value (or equivalent ``bytes``) returned
            by a Go-exported function, containing a UTF-8 encoded JSON array.

    Returns:
        A Python ``list`` parsed from the JSON content of ``cstr``.

    Raises:
        json.JSONDecodeError: If ``cstr`` does not contain valid JSON.
        UnicodeDecodeError: If the byte sequence in ``cstr`` is not valid UTF-8.
    """
    s = self.from_go_string(cstr)

    return json.loads(s)