我ModelBase
为我的基于 SQLAlchemy 的应用程序创建了一个类。
为了节省我不得不输入的繁琐工作,id = Column(Integer, primary_key=True)
我在我的ModelBase
班级中提供了该属性作为默认 ID。
Base = declarative_base()
class ModelBase(AbstractConcreteBase, Base):
id = Column(Integer, primary_key=True)
def __init__(self, *args, **kwargs):
""" constructor """
# do some nifty magic, keyword cleaning on kwargs, as we might have data
# coming from json input
super(ModelBase,self).__init__(*args, **kwargs)
def do_sth_else(self):
""" some shared fundamental logic, json (de)serialization """
我可能不应该这样做,因为现在所有类都有一个 id 整数字段。事实证明,我想在某些模型上使用复合键,但我仍然希望将模型类的默认值id
作为主键。所以我决定写一个mixin类,提供不同的ModelBase类。
Base = declarative_base()
class IDMixin(object):
id = Column(Integer, primary_key=True)
class AbstractModelBase(object):
def __init__(self, json_data='', *args, **kwargs):
""" same constructor """
super(AbstractModelBase,self).__init__(*args, **kwargs)
def do_sth_else(self):
""" same shared fundamental logic """
class ModelBase(AbstractConcreteBase, Base, IDMixin, AbstractModelBase):
""" new model base class """
class NoIDModelBase(AbstractConcreteBase, Base, AbstractModelBase):
""" new model base class """
然而,用关键字字典实例化这个类给了我一个痛苦的堆栈跟踪:
$ ./manage.py test # no, not django ;)
# Test 1 of 30
===============
Traceback (most recent call last):
File "./manage.py", line 40, in <module>
execute_command(sys.argv)
File "./manage.py", line 36, in execute_command
cmd(argv[2:])
File "./management/test.py", line 362, in handle
DBtestRunner(verbosity=args.verbosity).run(tests)
File "./management/test.py", line 172, in run
setUpDB(t)
File "./management/test.py", line 134, in setUpDB
instance = model_class(**d) ### instantiation occurs here ###
File "<string>", line 2, in __init__
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/orm/instrumentation.py", line 310, in _new_state_if_none
state = self._state_constructor(instance, self)
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/util/langhelpers.py", line 582, in __get__
obj.__dict__[self.__name__] = result = self.fget(obj)
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/orm/instrumentation.py", line 145, in _state_constructor
self.dispatch.first_init(self, self.class_)
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/event.py", line 409, in __call__
fn(*args, **kw)
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/orm/mapper.py", line 2197, in _event_on_first_init
configure_mappers()
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/orm/mapper.py", line 2123, in configure_mappers
_call_configured.dispatch.after_configured()
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/event.py", line 372, in __call__
fn(*args, **kw)
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/orm/events.py", line 489, in wrap
wrapped_fn(*arg, **kw)
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/ext/declarative/base.py", line 51, in go
cls.__declare_last__()
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/ext/declarative/api.py", line 347, in __declare_last__
cls.__mapper__ = m = mapper(cls, pjoin, polymorphic_on=pjoin.c.type)
File "/path/to/code/refactoring/.env/lib/python2.7/site-packages/sqlalchemy/util/_collections.py", line 172, in __getattr__
raise AttributeError(key)
AttributeError: type
在这个方法中调用模型类的构造函数:
def setUpDB(test):
test.db_engine = create_engine('sqlite:///:memory:', convert_unicode=True)
session_factory.configure(bind=test.db_engine)
db_session = scoped_session(session_factory)
ModelBase.metadata.create_all(test.db_engine)
test.client = Client(app, Response)
if (hasattr(test, "fixtures")):
# load fixtures from json
fixtures = test.fixtures
if isinstance(fixtures, basestring):
fixtures = (fixtures, )
elif isinstance(fixtures, collections.Iterable):
pass
else:
raise Exception(
"fixtures attribute needs to be string or iterable of strings")
for fixture in fixtures:
try:
with open(fixture, 'r') as f:
fixture = json.loads(f.read())
# apply fixture to database
for entry in fixture:
model = entry["model"]
# import the module containing the Model class
model_module = importlib.import_module(
model[:model.rfind(".")])
# get the model class
model_class = getattr(
model_module, model[model.rfind(".") + 1:])
# create an instance of the model class for
# each entry in "data"
data = entry["data"]
for d in data:
instance = model_class(**d)
instance.save()
except IOError as e:
print "Could not load Fixture!\n"
print e
sys.exit(1)