def __call__(self, p_struct, *args):
py_object_addr = getattr(p_struct.contents, self._py_object_field)
py_object = ctypes.cast(py_object_addr, ctypes.POINTER(ctypes.py_object)).contents.value
return self._fn(py_object, *args)
python类py_object()的实例源码
def __init__(self):
self.struct = self.CStructure()
self.this = ctypes.pointer(self.struct)
if self._pyobject_:
# save pointer to python object
setattr(self.struct, self._pyobject_, ctypes.cast(ctypes.pointer(ctypes.py_object(self)), ctypes.c_void_p))
self.override_vtable()
def _async_raise(self, tid, exctype):
"""raises the exception, performs cleanup if needed"""
tid = ctypes.c_long(tid)
if not inspect.isclass(exctype):
exctype = type(exctype)
res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
if res == 0:
log.error("invalid thread id")
elif res != 1:
"""if it returns a number greater than one, you're in trouble,
and you should call it again with exc=NULL to revert the effect"""
ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
log.error("PyThreadState_SetAsyncExc failed")
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.pydll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def stop(cls, task_name):
if task_name not in cls._tasks:
raise Exception("task {} is not exists".format(task_name))
tid = cls._tasks[task_name]
exc_type = SystemExit
tid = ctypes.c_long(tid)
res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exc_type))
if res == 0:
raise ValueError("invalid thread id?the task may be already stop")
elif res != 1:
# """if it returns a number greater than one, you're in trouble,
# and you should call it again with exc=NULL to revert the effect"""
ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
raise SystemError("PyThreadState_SetAsyncExc failed")
del cls._tasks[task_name]
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def dd_DDMatrix2Poly(matrixptr):
error = ctypes.c_int()
polyhedraptr = libcdd.dd_DDMatrix2Poly(matrixptr, ctypes.byref(error))
# Return None on error.
# The error values are enums, so they aren't exposed.
if error.value != DD_NO_ERRORS:
# Dump out the errors to stderr
libcdd.dd_WriteErrorMessages(
ctypes.pythonapi.PyFile_AsFile(ctypes.py_object(sys.stdout)),
error)
dd_FreePolyhedra(polyhedraptr)
return None
return polyhedraptr
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.pydll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def make_save_locals_impl():
"""
Factory for the 'save_locals_impl' method. This may seem like a complicated pattern but it is essential that the method is created at
module load time. Inner imports after module load time would cause an occasional debugger deadlock due to the importer lock and debugger
lock being taken in different order in different threads.
"""
try:
if '__pypy__' in sys.builtin_module_names:
import __pypy__ # @UnresolvedImport
pypy_locals_to_fast = __pypy__.locals_to_fast
except:
pass
else:
if '__pypy__' in sys.builtin_module_names:
def save_locals_pypy_impl(frame):
pypy_locals_to_fast(frame)
return save_locals_pypy_impl
try:
import ctypes
locals_to_fast = ctypes.pythonapi.PyFrame_LocalsToFast
except:
pass
else:
def save_locals_ctypes_impl(frame):
locals_to_fast(ctypes.py_object(frame), ctypes.c_int(0))
return save_locals_ctypes_impl
return None
def let_variable(self, frame_id, var_name, expression_value):
""" Let a frame's var with a value by building then eval a let
expression with breakoints disabled.
"""
breakpoints_backup = IKBreakpoint.backup_breakpoints_state()
IKBreakpoint.disable_all_breakpoints()
let_expression = "%s=%s" % (var_name, expression_value,)
eval_frame = ctypes.cast(frame_id, ctypes.py_object).value
global_vars = eval_frame.f_globals
local_vars = eval_frame.f_locals
try:
exec(let_expression, global_vars, local_vars)
error_message=""
except Exception as e:
t, result = sys.exc_info()[:2]
if isinstance(t, str):
result_type = t
else:
result_type = str(t.__name__)
error_message = "%s: %s" % (result_type, result,)
IKBreakpoint.restore_breakpoints_state(breakpoints_backup)
_logger.e_debug("let_variable(%s) => %s",
let_expression,
error_message or 'succeed')
return error_message
test_types_basic.py 文件源码
项目:aws-lambda-redshift-copy
作者: christianhxc
项目源码
文件源码
阅读 36
收藏 0
点赞 0
评论 0
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def __init__(self, vlc_instance):
# Cast this container to a c pointer to use in the callbacks
self._opaque = ctypes.cast(ctypes.pointer(
ctypes.py_object(self)), ctypes.c_void_p)
# Create the vlc callbacks, these will in turn call the methods defined
# in this container
self.media = vlc_instance.media_new_callbacks(
cb.CALLBACKS["read"],
cb.CALLBACKS["open"],
cb.CALLBACKS["seek"],
cb.CALLBACKS["close"],
self._opaque
)