Skip to content

interop

Implementation of reference counting classes for external resources accessed via interoperability software such as cffi.

Classes:

  • CffiNativeHandle

    Reference counting wrapper class for CFFI pointers.

  • CffiWrapperFactory

    A class that creates custom python wrappers based on the type identifier of the external pointer being wrapped.

  • DeletableCffiNativeHandle

    Reference counting wrapper class for CFFI pointers.

  • GenericWrapper

    A pass-through wrapper for python objects that are ready for C interop. "bytes" can be passed as C 'char*'.

  • OwningCffiNativeHandle

    Reference counting wrapper class for CFFI pointers that own and already manage the native memory.

Functions:

Attributes:

  • CffiData (TypeAlias) –

    A dummy type to use in type hints for limited documentation purposes

CffiData module-attribute

CffiData: TypeAlias = Any

A dummy type to use in type hints for limited documentation purposes

FFI.CData is a type, but it seems it cannot be used in type hinting.

CffiNativeHandle

CffiNativeHandle(
    handle: CffiData,
    type_id: Optional[str] = None,
    prior_ref_count: int = 0,
)

Bases: NativeHandle

Reference counting wrapper class for CFFI pointers.

This class is originally inspired from a class with a similar purpose in C#. See https://github.com/rdotnet/dynamic-interop-dll

Say you have a C API as follows:

  • void* create_some_object();
  • dispose_of_some_object(void* obj);

and accessing it using Python and CFFI. Users would use the calllib function:

from cffi import FFI

ffi = FFI()

# cdef() expects a single string declaring the C types, functions and
# globals needed to use the shared object. It must be in valid C syntax.
ffi.cdef('''
    void* create_some_object();
    dispose_of_some_object(void* obj);
''')
mydll_so = ffi.dlopen("/path/to/mydll.so", ffi.RTLD_LAZY)
cffi_void_ptr = mydll_so.create_some_object()

at some point when done you need to dispose of it to clear native memory:

mydll_so.dispose_of_some_object(cffi_void_ptr)

In practice in real systems one quickly ends up with cases where it is unclear when to dispose of the object. If you call the dispose_of_some_object function more than once, or too soon, you quickly crash the program, or possibly worse outcomes with numeric non-sense. CffiNativeHandle is designed to alleviate this headache by using native reference counting of handle classes to reliably dispose of objects.

Attributes:

  • _handle (object) –

    The handle (e.g. cffi pointer) to the native resource.

  • _type_id (Optional[str]) –

    An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.

  • _finalizing (bool) –

    a flag telling whether this object is in its deletion phase. This has a use in some advanced cases with reverse callback, possibly not relevant in Python.

Parameters:

  • handle (object) –

    The handle (e.g. cffi pointer) to the native resource.

  • type_id (Optional[str], default: None ) –

    An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.

  • prior_ref_count (int, default: 0 ) –

    the initial reference count. Default 0 if this NativeHandle is sole responsible for the lifecycle of the resource.

Methods:

  • add_ref

    Manually increment the reference count.

  • decrement_ref

    Manually increment the reference count.

  • dispose

    Disposing of the object pointed to by the CFFI pointer (handle) if the reference counts allows it.

  • get_handle

    Gets the underlying low-level CFFI handle this object wraps.

  • release

    Manually decrements the reference counter. Triggers disposal if reference count is down to zero.

Attributes:

  • disposed (bool) –

    Has the native object and memory already been disposed of.

  • is_invalid (bool) –

    Is the underlying handle valid? In practice synonym with the disposed attribute.

  • obj (Any) –

    Return the object pointed to (cffi object).

  • ptr (CffiData) –

    Return the pointer (cffi object).

  • reference_count (int) –

    Get the current reference count.

  • type_id (Optional[str]) –

    Return an optional type identifier for the underlying native type.

Source code in src/refcount/interop.py
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def __init__(self, handle: "CffiData", type_id: Optional[str] = None, prior_ref_count: int = 0):
    """Initialize a reference counter for a resource handle, with an initial reference count.

    Args:
        handle (object): The handle (e.g. cffi pointer) to the native resource.
        type_id (Optional[str]): An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.
        prior_ref_count (int): the initial reference count. Default 0 if this NativeHandle is sole responsible for the lifecycle of the resource.
    """
    super().__init__(handle, prior_ref_count)
    # TODO checks on handle
    self._type_id = type_id
    self._finalizing: bool = False
    self._handle: CffiData = None
    if handle is None:
        return  # defer setting the handle to the inheritor.
    self._set_handle(handle, prior_ref_count)

disposed property

disposed: bool

Has the native object and memory already been disposed of.

Returns:

  • bool

    The underlying native handle has been disposed of from this wrapper

is_invalid property

is_invalid: bool

Is the underlying handle valid? In practice synonym with the disposed attribute.

Returns:

  • bool

    True if this handle is valid

obj property

obj: Any

Return the object pointed to (cffi object).

ptr property

ptr: CffiData

Return the pointer (cffi object).

reference_count property

reference_count: int

Get the current reference count.

type_id property

type_id: Optional[str]

Return an optional type identifier for the underlying native type.

This can be in practice useful to be more transparent about the underlying type obtained via a C API with opaque pointers (i.e. void*)

Returns:

add_ref

add_ref() -> None

Manually increment the reference count.

Users usually have no need to call this method. They may have to if they manage cases where one native handle wrapper uses another wrapper (and its underlying resource).

Source code in src/refcount/base.py
26
27
28
29
30
31
32
def add_ref(self) -> None:
    """Manually increment the reference count.

    Users usually have no need to call this method. They may have to if they
    manage cases where one native handle wrapper uses another wrapper (and its underlying resource).
    """
    self._ref_count = self._ref_count + 1

decrement_ref

decrement_ref() -> None

Manually increment the reference count.

Users usually have no need to call this method. They may have to if they manage cases where one native handle wrapper uses another wrapper (and its underlying resource).

Source code in src/refcount/base.py
34
35
36
37
38
39
40
def decrement_ref(self) -> None:
    """Manually increment the reference count.

    Users usually have no need to call this method. They may have to if they
    manage cases where one native handle wrapper uses another wrapper (and its underlying resource).
    """
    self._ref_count = self._ref_count - 1

dispose

dispose() -> None

Disposing of the object pointed to by the CFFI pointer (handle) if the reference counts allows it.

Source code in src/refcount/interop.py
196
197
198
def dispose(self) -> None:
    """Disposing of the object pointed to by the CFFI pointer (handle) if the reference counts allows it."""
    self.__dispose_impl(True)

get_handle

get_handle() -> Union[CffiData, None]

Gets the underlying low-level CFFI handle this object wraps.

Returns:

Source code in src/refcount/interop.py
136
137
138
139
140
141
142
def get_handle(self) -> Union["CffiData", None]:
    """Gets the underlying low-level CFFI handle this object wraps.

    Returns:
        (Union[CffiData, None]): CFFI handle or None
    """
    return self._handle

release

release() -> None

Manually decrements the reference counter. Triggers disposal if reference count is down to zero.

Source code in src/refcount/interop.py
200
201
202
def release(self) -> None:
    """Manually decrements the reference counter. Triggers disposal if reference count is down to zero."""
    self.__dispose_impl(True)

CffiWrapperFactory

CffiWrapperFactory(
    api_type_wrapper: Dict[str, Any],
    strict_wrapping: bool = False,
)

A class that creates custom python wrappers based on the type identifier of the external pointer being wrapped.

Parameters:

  • api_type_wrapper (Dict[str, Any]) –

    dictionary, mapping from type identifiers to callables, class constructors

  • strict_wrapping (bool, default: False ) –

    If true, type identifiers passed at wrapper creation time create_wrapper must be known or exceptions are raised. If False, it falls back on creating generic wrappers. Defaults to False.

Methods:

  • create_wrapper

    Create a CffiNativeHandle wrapper around an object, if this object is a CFFI pointer.

Source code in src/refcount/interop.py
458
459
460
461
462
463
464
465
466
467
def __init__(self, api_type_wrapper: Dict[str, Any], strict_wrapping: bool = False) -> None:
    """A class that creates custom python wrappers based on the type identifier of the external pointer being wrapped.

    Args:
        api_type_wrapper (Dict[str,Any]): dictionary, mapping from type identifiers to callables, class constructors
        strict_wrapping (bool, optional): If true, type identifiers passed at wrapper creation time `create_wrapper`
            must be known or exceptions are raised. If False, it falls back on creating generic wrappers. Defaults to False.
    """
    self._strict_wrapping = strict_wrapping
    self._api_type_wrapper = api_type_wrapper

create_wrapper

create_wrapper(
    obj: Any,
    type_id: str,
    release_native: Optional[Callable[[CffiData], None]],
) -> CffiNativeHandle

Create a CffiNativeHandle wrapper around an object, if this object is a CFFI pointer.

Parameters:

  • obj (Union[CffiData, Any]) –

    An object, which will be wrapped if this is a CFFI pointer, i.e. an instance of CffiData

  • type_id (Optional[str]) –

    An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.

  • release_native (Callable[[CffiData], None]) –

    function to call on deleting this wrapper. The function should have one argument accepting the object handle.

Raises:

  • ValueError

    Missing type_id

  • ValueError

    If this object is in strict mode, and type_id is not known in the mapping

  • NotImplementedError

    type_id is known, but mapping to None (wrapper not yet implemented)

  • TypeError

    The function to create the wrapper does not accept any argument.

Returns:

Source code in src/refcount/interop.py
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
def create_wrapper(
    self,
    obj: Any,
    type_id: str,
    release_native: Optional[Callable[["CffiData"], None]],
) -> "CffiNativeHandle":
    """Create a CffiNativeHandle wrapper around an object, if this object is a CFFI pointer.

    Args:
        obj (Union[CffiData,Any]): An object, which will be wrapped if this is a CFFI pointer, i.e. an instance of `CffiData`
        type_id (Optional[str]): An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.
        release_native (Callable[[CffiData],None]): function to call on deleting this wrapper. The function should have one argument accepting the object handle.

    Raises:
        ValueError: Missing type_id
        ValueError: If this object is in strict mode, and `type_id` is not known in the mapping
        NotImplementedError: `type_id` is known, but mapping to None (wrapper not yet implemented)
        TypeError: The function to create the wrapper does not accept any argument.

    Returns:
        CffiNativeHandle: cffi wrapper
    """
    from inspect import signature

    # from typing import get_type_hints
    if type_id is None:
        raise ValueError("Type ID provided cannot be None")
    if type_id not in self._api_type_wrapper:
        if self._strict_wrapping:
            raise ValueError(f"Type ID {type_id} is unknown")
        return wrap_cffi_native_handle(obj, type_id, release_native)
    wrapper_type = self._api_type_wrapper[type_id]
    if wrapper_type is None:
        if self._strict_wrapping:
            raise NotImplementedError(
                f"Python object wrapper for foreign type ID {wrapper_type} is not yet implemented",
            )
        return wrap_cffi_native_handle(obj, type_id, release_native)
    s = signature(wrapper_type)
    n = len(s.parameters)
    parameters = [v for k, v in s.parameters.items()]
    # [<Parameter "handle: Any">, <Parameter "release_native: Callable[[Any], NoneType]">, <Parameter "type_id: Optional[str] = None">, <Parameter "prior_ref_count: int = 0">]
    if n == 0:
        raise TypeError(
            f"Wrapper class '{wrapper_type.__name__}' has no constructor arguments; at least one is required",
        )
    if n == 1:
        return wrapper_type(obj)
    if n == 2:  # noqa: PLR2004
        if release_native is None:
            raise ValueError(
                f"Wrapper class '{wrapper_type.__name__}' has two constructor arguments; the argument 'release_native' cannot be None",
            )
        return wrapper_type(obj, release_native)
    if n == 3:  # noqa: PLR2004
        if release_native is None:
            raise ValueError(
                f"Wrapper class '{type(wrapper_type)}' has three constructor arguments; the argument 'release_native' cannot be None",
            )
        return wrapper_type(obj, release_native, type_id)
    if n == 4:  # noqa: PLR2004
        p = parameters[3]
        # constructor = wrapper_type.__init__
        # type_hints = get_type_hints(constructor)
        param_type = p.annotation
        if param_type is not int:
            raise TypeError(
                f"Wrapper class '{type(wrapper_type)}' has four constructor arguments; the last argument 'prior_ref_count' must be an integer",
            )
        if parameters[3].default == parameters[3].empty:
            raise ValueError(
                f"Wrapper class '{type(wrapper_type)}' has four constructor arguments; the last argument 'prior_ref_count' must have a default value",
            )
        return wrapper_type(obj, release_native, type_id)
    raise NotImplementedError(
        f"Wrapper class '{wrapper_type.__name__}' has more than 4 arguments; this is not supported",
    )

DeletableCffiNativeHandle

DeletableCffiNativeHandle(
    handle: CffiData,
    release_native: Optional[Callable[[CffiData], None]],
    type_id: Optional[str] = None,
    prior_ref_count: int = 0,
)

Bases: CffiNativeHandle

Reference counting wrapper class for CFFI pointers.

Attributes:

  • _handle (object) –

    The handle (e.g. cffi pointer) to the native resource.

  • _type_id (Optional[str]) –

    An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.

  • _finalizing (bool) –

    a flag telling whether this object is in its deletion phase. This has a use in some advanced cases with reverse callback, possibly not relevant in Python.

  • _release_native (Callable[[CffiData], None]) –

    function to call on deleting this wrapper. The function should have one argument accepting the object _handle.

Parameters:

  • handle (CffiData) –

    The handle (expected cffi pointer) to the native resource.

  • release_native (Callable[[CffiData], None]) –

    function to call on deleting this wrapper. The function should have one argument accepting the object handle.

  • type_id (str, default: None ) –

    [description]. An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation. Defaults to None.

  • prior_ref_count (int, default: 0 ) –

    [description]. The initial reference count. Defaults to 0 if this NativeHandle is sole responsible for the lifecycle of the resource.

Methods:

  • add_ref

    Manually increment the reference count.

  • decrement_ref

    Manually increment the reference count.

  • dispose

    Disposing of the object pointed to by the CFFI pointer (handle) if the reference counts allows it.

  • get_handle

    Gets the underlying low-level CFFI handle this object wraps.

  • release

    Manually decrements the reference counter. Triggers disposal if reference count is down to zero.

Attributes:

  • disposed (bool) –

    Has the native object and memory already been disposed of.

  • is_invalid (bool) –

    Is the underlying handle valid? In practice synonym with the disposed attribute.

  • obj (Any) –

    Return the object pointed to (cffi object).

  • ptr (CffiData) –

    Return the pointer (cffi object).

  • reference_count (int) –

    Get the current reference count.

  • type_id (Optional[str]) –

    Return an optional type identifier for the underlying native type.

Source code in src/refcount/interop.py
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def __init__(
    self,
    handle: "CffiData",
    release_native: Optional[Callable[["CffiData"], None]],
    type_id: Optional[str] = None,
    prior_ref_count: int = 0,
):
    """New reference counter for a CFFI resource handle.

    Args:
        handle (CffiData): The handle (expected cffi pointer) to the native resource.
        release_native (Callable[[CffiData],None]): function to call on deleting this wrapper. The function should have one argument accepting the object handle.
        type_id (str, optional): [description]. An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation. Defaults to None.
        prior_ref_count (int, optional): [description]. The initial reference count. Defaults to 0 if this NativeHandle is sole responsible for the lifecycle of the resource.
    """
    super().__init__(
        handle,
        type_id,
        prior_ref_count,
    )
    self._release_native = release_native
    self._set_handle(handle, prior_ref_count)

disposed property

disposed: bool

Has the native object and memory already been disposed of.

Returns:

  • bool

    The underlying native handle has been disposed of from this wrapper

is_invalid property

is_invalid: bool

Is the underlying handle valid? In practice synonym with the disposed attribute.

Returns:

  • bool

    True if this handle is valid

obj property

obj: Any

Return the object pointed to (cffi object).

ptr property

ptr: CffiData

Return the pointer (cffi object).

reference_count property

reference_count: int

Get the current reference count.

type_id property

type_id: Optional[str]

Return an optional type identifier for the underlying native type.

This can be in practice useful to be more transparent about the underlying type obtained via a C API with opaque pointers (i.e. void*)

Returns:

add_ref

add_ref() -> None

Manually increment the reference count.

Users usually have no need to call this method. They may have to if they manage cases where one native handle wrapper uses another wrapper (and its underlying resource).

Source code in src/refcount/base.py
26
27
28
29
30
31
32
def add_ref(self) -> None:
    """Manually increment the reference count.

    Users usually have no need to call this method. They may have to if they
    manage cases where one native handle wrapper uses another wrapper (and its underlying resource).
    """
    self._ref_count = self._ref_count + 1

decrement_ref

decrement_ref() -> None

Manually increment the reference count.

Users usually have no need to call this method. They may have to if they manage cases where one native handle wrapper uses another wrapper (and its underlying resource).

Source code in src/refcount/base.py
34
35
36
37
38
39
40
def decrement_ref(self) -> None:
    """Manually increment the reference count.

    Users usually have no need to call this method. They may have to if they
    manage cases where one native handle wrapper uses another wrapper (and its underlying resource).
    """
    self._ref_count = self._ref_count - 1

dispose

dispose() -> None

Disposing of the object pointed to by the CFFI pointer (handle) if the reference counts allows it.

Source code in src/refcount/interop.py
196
197
198
def dispose(self) -> None:
    """Disposing of the object pointed to by the CFFI pointer (handle) if the reference counts allows it."""
    self.__dispose_impl(True)

get_handle

get_handle() -> Union[CffiData, None]

Gets the underlying low-level CFFI handle this object wraps.

Returns:

Source code in src/refcount/interop.py
136
137
138
139
140
141
142
def get_handle(self) -> Union["CffiData", None]:
    """Gets the underlying low-level CFFI handle this object wraps.

    Returns:
        (Union[CffiData, None]): CFFI handle or None
    """
    return self._handle

release

release() -> None

Manually decrements the reference counter. Triggers disposal if reference count is down to zero.

Source code in src/refcount/interop.py
200
201
202
def release(self) -> None:
    """Manually decrements the reference counter. Triggers disposal if reference count is down to zero."""
    self.__dispose_impl(True)

GenericWrapper

GenericWrapper(handle: CffiData)

A pass-through wrapper for python objects that are ready for C interop. "bytes" can be passed as C 'char*'.

This is mostly a facility to generate glue code more easily

Attributes:

  • ptr (CffiData) –

    The wrapped python object that is ready for C interop.

Source code in src/refcount/interop.py
394
395
396
def __init__(self, handle: "CffiData"):
    """A pass-through wrapper for python objects that are ready for C interop. "bytes" can be passed as C 'char*'."""
    self._handle = handle

ptr property

ptr: CffiData

The wrapped python object that is ready for C interop.

OwningCffiNativeHandle

OwningCffiNativeHandle(
    handle: CffiData,
    type_id: Optional[str] = None,
    prior_ref_count: int = 0,
)

Bases: CffiNativeHandle

Reference counting wrapper class for CFFI pointers that own and already manage the native memory.

Attributes:

  • _handle (object) –

    The handle (e.g. cffi pointer) to the native resource.

  • _type_id (Optional[str]) –

    An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.

  • _finalizing (bool) –

    a flag telling whether this object is in its deletion phase. This has a use in some advanced cases with reverse callback, possibly not relevant in Python.

Parameters:

  • handle (CffiData) –

    The handle (expected cffi pointer) to the native resource.

  • type_id (str, default: None ) –

    [description]. An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation. Defaults to None.

  • prior_ref_count (int, default: 0 ) –

    [description]. The initial reference count. Defaults to 0 if this NativeHandle is sole responsible for the lifecycle of the resource.

Methods:

  • add_ref

    Manually increment the reference count.

  • decrement_ref

    Manually increment the reference count.

  • dispose

    Disposing of the object pointed to by the CFFI pointer (handle) if the reference counts allows it.

  • get_handle

    Gets the underlying low-level CFFI handle this object wraps.

  • release

    Manually decrements the reference counter. Triggers disposal if reference count is down to zero.

Attributes:

  • disposed (bool) –

    Has the native object and memory already been disposed of.

  • is_invalid (bool) –

    Is the underlying handle valid? In practice synonym with the disposed attribute.

  • obj (Any) –

    Return the object pointed to (cffi object).

  • ptr (CffiData) –

    Return the pointer (cffi object).

  • reference_count (int) –

    Get the current reference count.

  • type_id (Optional[str]) –

    Return an optional type identifier for the underlying native type.

Source code in src/refcount/interop.py
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
def __init__(
    self,
    handle: "CffiData",
    type_id: Optional[str] = None,
    prior_ref_count: int = 0,
):
    """Reference counting wrapper class for CFFI pointers that own and already manage the native memory.

    Args:
        handle (CffiData): The handle (expected cffi pointer) to the native resource.
        type_id (str, optional): [description]. An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation. Defaults to None.
        prior_ref_count (int, optional): [description]. The initial reference count. Defaults to 0 if this NativeHandle is sole responsible for the lifecycle of the resource.
    """
    super().__init__(
        handle,
        type_id,
        prior_ref_count,
    )
    self._set_handle(handle, prior_ref_count)

disposed property

disposed: bool

Has the native object and memory already been disposed of.

Returns:

  • bool

    The underlying native handle has been disposed of from this wrapper

is_invalid property

is_invalid: bool

Is the underlying handle valid? In practice synonym with the disposed attribute.

Returns:

  • bool

    True if this handle is valid

obj property

obj: Any

Return the object pointed to (cffi object).

ptr property

ptr: CffiData

Return the pointer (cffi object).

reference_count property

reference_count: int

Get the current reference count.

type_id property

type_id: Optional[str]

Return an optional type identifier for the underlying native type.

This can be in practice useful to be more transparent about the underlying type obtained via a C API with opaque pointers (i.e. void*)

Returns:

add_ref

add_ref() -> None

Manually increment the reference count.

Users usually have no need to call this method. They may have to if they manage cases where one native handle wrapper uses another wrapper (and its underlying resource).

Source code in src/refcount/base.py
26
27
28
29
30
31
32
def add_ref(self) -> None:
    """Manually increment the reference count.

    Users usually have no need to call this method. They may have to if they
    manage cases where one native handle wrapper uses another wrapper (and its underlying resource).
    """
    self._ref_count = self._ref_count + 1

decrement_ref

decrement_ref() -> None

Manually increment the reference count.

Users usually have no need to call this method. They may have to if they manage cases where one native handle wrapper uses another wrapper (and its underlying resource).

Source code in src/refcount/base.py
34
35
36
37
38
39
40
def decrement_ref(self) -> None:
    """Manually increment the reference count.

    Users usually have no need to call this method. They may have to if they
    manage cases where one native handle wrapper uses another wrapper (and its underlying resource).
    """
    self._ref_count = self._ref_count - 1

dispose

dispose() -> None

Disposing of the object pointed to by the CFFI pointer (handle) if the reference counts allows it.

Source code in src/refcount/interop.py
196
197
198
def dispose(self) -> None:
    """Disposing of the object pointed to by the CFFI pointer (handle) if the reference counts allows it."""
    self.__dispose_impl(True)

get_handle

get_handle() -> Union[CffiData, None]

Gets the underlying low-level CFFI handle this object wraps.

Returns:

Source code in src/refcount/interop.py
136
137
138
139
140
141
142
def get_handle(self) -> Union["CffiData", None]:
    """Gets the underlying low-level CFFI handle this object wraps.

    Returns:
        (Union[CffiData, None]): CFFI handle or None
    """
    return self._handle

release

release() -> None

Manually decrements the reference counter. Triggers disposal if reference count is down to zero.

Source code in src/refcount/interop.py
200
201
202
def release(self) -> None:
    """Manually decrements the reference counter. Triggers disposal if reference count is down to zero."""
    self.__dispose_impl(True)

cffi_arg_error_external_obj_type

cffi_arg_error_external_obj_type(
    x: Any, expected_type_id: str
) -> str

Build an error message that an unexpected object is in lieu of an expected refcount external ref object.

Parameters:

  • x (object) –

    object passed as an argument to a function but with an unexpected type or type id.

  • expected_type_id (Optional[str]) –

    Expected identifier for the type of underlying resource being wrapped.

Returns (str): the error message

Source code in src/refcount/interop.py
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def cffi_arg_error_external_obj_type(x: Any, expected_type_id: str) -> str:
    """Build an error message that an unexpected object is in lieu of an expected refcount external ref object.

    Args:
        x (object): object passed as an argument to a function but with an unexpected type or type id.
        expected_type_id (Optional[str]): Expected identifier for the type of underlying resource being wrapped.

    Returns (str): the error message
    """
    if x is None:
        return "Expected a 'CffiNativeHandle' but instead got 'None'"
    if not is_cffi_native_handle(x):
        return f"Expected a 'CffiNativeHandle' but instead got object of type '{type(x)!s}'"
    return f"Expected a 'CffiNativeHandle' with underlying type id '{expected_type_id}' but instead got one with type id '{x.type_id}'"

is_cffi_native_handle

is_cffi_native_handle(x: Any, type_id: str = '') -> bool

Checks whether an object is a ref counting wrapper around a CFFI pointer.

Parameters:

  • x (object) –

    object to test, presumed to be an instance of CffiNativeHandle

  • type_id (Optional[str], default: '' ) –

    Optional identifier for the type of underlying resource being wrapped.

Source code in src/refcount/interop.py
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def is_cffi_native_handle(x: Any, type_id: str = "") -> bool:
    """Checks whether an object is a ref counting wrapper around a CFFI pointer.

    Args:
        x (object): object to test, presumed to be an instance of `CffiNativeHandle`
        type_id (Optional[str]): Optional identifier for the type of underlying resource being wrapped.
    """
    if x is None:
        return False
    if not isinstance(x, CffiNativeHandle):
        return False
    if type_id is None or type_id == "":
        return True
    return x.type_id == type_id

type_error_cffi

type_error_cffi(
    x: Union[CffiNativeHandle, Any], expected_type: str
) -> str

DEPRECATED.

This function is deprecated; may still be in use in 'uchronia'. Use cffi_arg_error_external_obj_type instead.

Build an error message for situations where a cffi pointer handler is not that, or not of the expected type

Parameters:

  • x (Union[CffiNativeHandle, Any]) –

    actual object that is not of the expected type or underlying type for the external pointer.

  • expected_type (str) –

    underlying type expected for the CFFI pointer handler

Returns:

  • str ( str ) –

    error message that the caller can use to report the issue

Source code in src/refcount/interop.py
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
def type_error_cffi(x: Union[CffiNativeHandle, Any], expected_type: str) -> str:
    """DEPRECATED.

    This function is deprecated; may still be in use in 'uchronia'. Use `cffi_arg_error_external_obj_type` instead.

    Build an error message for situations where a cffi pointer handler is not that, or not of the expected type

    Args:
        x (Union[CffiNativeHandle, Any]): actual object that is not of the expected type or underlying type for the external pointer.
        expected_type (str): underlying type expected for the CFFI pointer handler

    Returns:
        str: error message that the caller can use to report the issue
    """
    return cffi_arg_error_external_obj_type(x, expected_type)

unwrap_cffi_native_handle

unwrap_cffi_native_handle(
    obj_wrapper: Any, stringent: bool = False
) -> Union[CffiData, Any, None]

Unwrap a reference counting wrapper and returns its CFFI pointer if it is found (wrapped or 'raw').

Parameters:

  • obj_wrapper (Any) –

    An object, which will be unwrapped if this is a CFFI pointer, i.e. an instance of CffiData

  • stringent (bool, default: False ) –

    [description]. if True an error is raised if obj_wrapper is neither None, a CffiNativeHandle nor an CffiData. Defaults to False.

Raises:

  • Exception

    A CFFI pointer could not be found in the object.

Returns:

  • Union[CffiData, Any, None]

    Union[CffiData,Any,None]: A CFFI pointer if it was found. Returns None or unchanged if not found, and stringent is equal to False. Exception otherwise.

Source code in src/refcount/interop.py
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
def unwrap_cffi_native_handle(
    obj_wrapper: Any,
    stringent: bool = False,
) -> Union["CffiData", Any, None]:
    """Unwrap a reference counting wrapper and returns its CFFI pointer if it is found (wrapped or 'raw').

    Args:
        obj_wrapper (Any): An object, which will be unwrapped if this is a CFFI pointer, i.e. an instance of `CffiData`
        stringent (bool, optional): [description]. if True an error is raised if obj_wrapper is neither None, a CffiNativeHandle nor an CffiData. Defaults to False.

    Raises:
        Exception: A CFFI pointer could not be found in the object.

    Returns:
        Union[CffiData,Any,None]: A CFFI pointer if it was found. Returns None or unchanged if not found, and stringent is equal to False. Exception otherwise.
    """
    # 2016-01-28 allowing null pointers, to unlock behavior of EstimateERRISParameters.
    # Reassess approach, even if other C API function will still catch the issue of null ptrs.
    if obj_wrapper is None:
        return None
    if isinstance(obj_wrapper, CffiNativeHandle):
        return obj_wrapper.get_handle()
    if isinstance(obj_wrapper, FFI.CData):
        return obj_wrapper
    if stringent:
        raise TypeError(
            "Argument is neither a CffiNativeHandle nor a CFFI external pointer",
        )
    return obj_wrapper

wrap_as_pointer_handle

wrap_as_pointer_handle(
    obj_wrapper: Any, stringent: bool = False
) -> Union[
    CffiNativeHandle, OwningCffiNativeHandle, GenericWrapper
]

Wrap an object, if need be, so that its C API pointer appears accessible via a 'ptr' property.

Parameters:

  • obj_wrapper (Any) –

    Object to wrap, if necessary

  • stringent (bool, default: False ) –

    Throws an exception if the input type is unhandled. Defaults to False.

Raises:

  • TypeError

    neither a CffiNativeHandle nor a CFFI external pointer, nor bytes

Returns:

Source code in src/refcount/interop.py
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
def wrap_as_pointer_handle(
    obj_wrapper: Any,
    stringent: bool = False,
) -> Union[CffiNativeHandle, OwningCffiNativeHandle, GenericWrapper]:
    """Wrap an object, if need be, so that its C API pointer appears accessible via a 'ptr' property.

    Args:
        obj_wrapper (Any): Object to wrap, if necessary
        stringent (bool, optional): Throws an exception if the input type is unhandled. Defaults to False.

    Raises:
        TypeError: neither a CffiNativeHandle nor a CFFI external pointer, nor bytes

    Returns:
        Union[CffiNativeHandle, OwningCffiNativeHandle, GenericWrapper, None]: wrapped object or None
    """
    # 2016-01-28 allowing null pointers, to unlock behavior of EstimateERRISParameters.
    # Reassess approach, even if other C API function will still catch the issue of null ptrs.
    if obj_wrapper is None:
        return GenericWrapper(None)
        # return GenericWrapper(FFI.NULL)  # Ended with kernel crashes and API call return, but unclear why
    if isinstance(obj_wrapper, CffiNativeHandle):
        return obj_wrapper
    if isinstance(obj_wrapper, FFI.CData):
        return OwningCffiNativeHandle(obj_wrapper)
    if isinstance(obj_wrapper, bytes):
        return GenericWrapper(obj_wrapper)
    if stringent:
        raise TypeError(
            "Argument is neither a CffiNativeHandle nor a CFFI external pointer, nor bytes",
        )
    return obj_wrapper

wrap_cffi_native_handle

wrap_cffi_native_handle(
    obj: Union[CffiData, Any],
    type_id: str = "",
    release_native: Optional[
        Callable[[CffiData], None]
    ] = None,
) -> Union[DeletableCffiNativeHandle, Any]

Create a reference counting wrapper around an object if this object is a CFFI pointer.

Parameters:

  • obj (Union[CffiData, Any]) –

    An object, which will be wrapped if this is a CFFI pointer, i.e. an instance of CffiData

  • release_native (Callable[[CffiData], None], default: None ) –

    function to call on deleting this wrapper. The function should have one argument accepting the object handle.

  • type_id (Optional[str], default: '' ) –

    An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.

Source code in src/refcount/interop.py
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
def wrap_cffi_native_handle(
    obj: Union["CffiData", Any],
    type_id: str = "",
    release_native: Optional[Callable[["CffiData"], None]] = None,
) -> Union[DeletableCffiNativeHandle, Any]:
    """Create a reference counting wrapper around an object if this object is a CFFI pointer.

    Args:
        obj (Union[CffiData,Any]): An object, which will be wrapped if this is a CFFI pointer, i.e. an instance of `CffiData`
        release_native (Callable[[CffiData],None]): function to call on deleting this wrapper. The function should have one argument accepting the object handle.
        type_id (Optional[str]): An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.
    """
    if isinstance(obj, FFI.CData):
        return DeletableCffiNativeHandle(
            obj,
            release_native=release_native,
            type_id=type_id,
        )
    return obj