Mixin Cython Class and SqlAlchemy
Annotation:
I have one cython class that represents a business unit. This class is declared in pure cython style.
In one project I need to map a business unit to a database. For this I would like to import a .pxd file and "map" it to SQLAlchemy.
Cython definition
Suppose the class Equipment. The class is defined by .pyx and the interface of the class in .pxd (because I need to import it in another module).
equipment.pxd
cdef class Equipment:
cdef readonly int x
cdef readonly str y
equipment.pyx
cdef class Equipment:
def __init__(self, int x, str y):
self.x = x
self.y = y
I put everything together and get the equipment.pyd file. So far so good. This file contains the business logic model and should not be changed.
Mapping
Then, in one application, I import a .pyd hardware and map it to SQLAlchemy.
from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.orm import mapper
from equipment import Equipment
metadata = MetaData()
# Table definition
equipment = Table(
'equipment', metadata,
Column('id', Integer, primary_key=True),
Column('x', Integer),
Column('y', String),
)
# Mapping the table definition with the class definition
mapper(Equipment, equipment)
TypeError: cannot set builtin / extension attributes of type "equipment.Equipment"
Indeed, SQLAlchemy tries to create Equipment.cx, Equipment.cy, ... which is not possible in Cython because it is not defined in .pxd ...
So how can I map the Cython class to SQLAlchemy?
Unsatisfying solution
If I define a hardware class in python mode in a .pyx file, it works because at the end it is just a "python" object in the cython class definition.
equipment.pyx
class Equipment:
def __init__(self, x, y):
self.x = x
self.y = y
But I am losing a lot of functionality, so I need a pure Kyoto.
Thank!: -)
- EDIT PART -
Semi-durable solution
Save the .pyx and .pxd files. Inherit from .pyd. Try to find a map.
mapping.py
from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.orm import mapper
from equipment import Equipment
metadata = MetaData()
# Table definition
equipment = Table(
'equipment', metadata,
Column('id', Integer, primary_key=True),
Column('x', Integer),
Column('y', String),
)
# Inherit Equipment to a mapped class
class EquipmentMapped(Equipment):
def __init__(self, x, y):
super(EquipmentMapped, self).__init__(x, y)
# Mapping the table definition with the class definition
mapper(EquipmentMapped, equipment)
from import mapping EquipmentMapped
e = EquipmentMapped (2, 3)
print ex
## This is empty!
To make it work, I have to define each attribute as a property!
equipment.pxd
cdef class Equipment:
cdef readonly int _x
cdef readonly str _y
equipment.pyx
cdef class Equipment:
def __init__(self, int x, str y):
self.x = x
self.y = y
property x:
def __get__(self):
return self._x
def __set__(self, x):
self._x = x
property y:
def __get__(self):
return self._y
def __set__(self, y):
self._y = y
This is not satisfying because: lazy_programmer_mode: I have a lot of changes in business logic ...: lazy_programmer_mode off:
source to share
I think the main problem is that when you call mapper
it does (among other things)
Equipment.x = ColumnProperty(...) # with some arguments
Equipment.y = ColumnProperty(...)
when ColumnProperty
is a sqlalchemy defined property, so when you do e.x = 5
, it may notice that the value has changed in all database related stuff.
Obviously it doesn't play well with the Cython class below, which you are trying to use to manage storage.
Personally, I suspect the only real answer is to define a wrapper class that contains both the Cython class and the sqlalchemy display class, and intercepts all attribute calls and method calls to keep them in sync. Below is a rough implementation that should work for simple cases. It is barely tested, so there are almost certainly bugs and corner cases that it misses. Caution!
def wrapper_class(cls):
# do this in a function so we can create it generically as needed
# for each cython class
class WrapperClass(object):
def __init__(self,*args,**kwargs):
# construct the held class using arguments provided
self._wrapped = cls(*args,**kwargs)
def __getattribute__(self,name):
# intercept all requests for attribute access.
wrapped = object.__getattribute__(self,"_wrapped")
update_from = wrapped
update_to = self
try:
o = getattr(wrapped,name)
except AttributeError:
# if we can't find it look in this class instead.
# This is reasonable, because there may be methods defined
# by sqlalchemy for example
update_from = self
update_to = wrapped
o = object.__getattribute__(self,name)
if callable(o):
return FunctionWrapper(o,update_from,update_to)
else:
return o
def __setattr__(self,name,value):
# intercept all attempt to write to attributes
# and set it in both this class and the wrapped Cython class
if name!="_wrapped":
try:
setattr(self._wrapped,name,value)
except AttributeError:
pass # ignore errors... maybe bad!
object.__setattr__(self,name,value)
return WrapperClass
class FunctionWrapper(object):
# a problem we have is if we call a member function.
# It possible that the member function may change something
# and thus we need to ensure that everything is updated appropriately
# afterwards
def __init__(self,func,update_from,update_to):
self.__func = func
self.__update_from = update_from
self.__update_to = update_to
def __call__(self,*args,**kwargs):
ret_val = self.__func(*args,**kwargs)
# for both Cython classes and sqlalchemy mappings
# all the relevant attributes exist in the class dictionary
for k in self.__update_from.__class__.__dict__.iterkeys():
if not k.startswith('__'): # ignore private stuff
try:
setattr(self.__update_to,k,getattr(self.__update_from,k))
except AttributeError:
# There may be legitmate cases when this fails
# (probably relating to sqlalchemy functions?)
# in this case, replace raise with pass
raise
return ret_val
To use it, you would do something like:
class EquipmentMapped(wrapper_class(Equipment)):
# you may well have to define __init__ here
# you'll have to check yourself, and see what sqlalchemy does...
pass
mapper(EquipmentMapped,equipment)
Please remember that this is a terrible workflow that basically just duplicates all of your data in two places and then desperately tries to sync.
Edit . The original version of this gave the OP's automated query line mechanism, but decided there was a lot of effort to do it manually (defining properties in a Cython class that only succeeds in overriding sqlalchemy mechanisms to track changes). Further testing confirms that it doesn't work. If you're wondering what not to do, check out the changelog!
source to share