我正在使用 factory_boy 创建测试装置。我有两个简单的工厂,由 SQLAlchemy 模型支持(如下简化)。
我希望能够多次调用
AddressFactory.create()
,并让它创建一个 Country
(如果它尚不存在),否则我希望它重新使用现有记录。
class CountryFactory(factory.Factory):
FACTORY_FOR = Country
cc = "US"
name = "United States"
class AddressFactory(factory.Factory):
FACTORY_FOR = Address
name = "Joe User"
city = "Seven Mile Beach"
country = factory.SubFactory(CountryFactory, cc="KY", name="Cayman Islands")
我的问题是:我如何设置这些工厂,以便factory_boy不会在每次创建地址时都尝试创建一个新的国家/地区?
在最新的factory-boy==2.3.1中可以添加FACTORY_DJANGO_GET_OR_CREATE
class CountryFactory(factory.django.DjangoModelFactory):
FACTORY_FOR = 'appname.Country'
FACTORY_DJANGO_GET_OR_CREATE = ('cc',)
cc = "US"
name = "United States"
假设cc字段是唯一标识符。
虽然基于 SQLAlchemy 的工厂没有
get_or_create
函数是对的,但如果要用作外键的对象已经存在,则可以迭代它们:
http://factoryboy.readthedocs.org/en/latest/recipes.html#choosing-from-a-populated-table
因此,可以想象,您可以通过使用惰性属性在工厂中组合一个解决方案,该属性首先检查数据库中是否存在该对象,如果存在,则使用此方法来迭代它们,但如果该对象不存在,则它首先调用
SubFactory
创建对象。
对于SqlAlchemy,你可以尝试这个。这也是缓存工厂:
class StaticFactory(factory.alchemy.SQLAlchemyModelFactory):):
__static_exclude = ('__static_exclude', '__static_cache',)
__static_cache = {}
@classmethod
def _create(cls, model_class, *args, **kwargs):
"""Helper for avoid duplicate factory"""
# Exclude static cache
cls._meta.exclude += cls.__static_exclude
_unique_key = None
# Get first unique keys from table. I'll be cache key.
for col in model_class.__table__.columns:
if any([col.primary_key, col.unique]):
_unique_key = kwargs.get(col.name)
if _unique_key:
break
_instance = cls.__static_cache.get(_unique_key)
if _instance:
return _instance
_session = cls._meta.sqlalchemy_session
with _session.no_autoflush:
obj = model_class(*args, **kwargs)
_session.add(obj)
cls.__static_cache.update({_unique_key: obj})
return obj
class LanguageFactory(StaticFactory):
class Meta:
model = Language
exclude = ('lang',)
另一个 hacky 解决方案是重写工厂的
create
方法,通过查询和缓存结果来搜索对象。
这个简单的示例没有对
**kwargs
进行过滤:
class StaticFactory(SQLAlchemyModelFactory):
counter = 0
cache = []
model = None
@classmethod
def create(cls, **kwargs):
if not cls.cache:
cls.cache = your_session.query(cls.model).all()
instance = cls.cache[cls.counter]
cls.counter = (cls.counter + 1) % len(cls.cache)
return instance
我们可以使用 factory.Iterator 方法使用现有的国家/地区实例创建一个新的地址实例
import factory, factory.django
from . import models
class CountryFactory(factory.Factory.DjangoModelFactory):
model = models.Country
cc = "US"
name = "United States"
class AddressFactory(factory.Factory.DjangoModelFactory):
model = models.Address
name = "Joe User"
city = "Seven Mile Beach"
country = factory.Iterator(models.Country.objects.all())
在这里,我们从数据库中访问了 Country 实例,并将其传递给 AddressFactory 的国家属性,后者使用数据库中已创建的国家实例创建了一个地址实例。
从版本 3.0.0 开始,SQLAlchemy 工厂支持
sqlalchemy_get_or_create
选项。正如文档所说,“在此列表中传递名称的字段将用于执行 Model.query.one_or_none() 或通常的 Session.add()”。
使用文档中的示例:
class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
class Meta:
model = User
sqlalchemy_session = session
sqlalchemy_get_or_create = ('username',)
username = 'john'
>>> User.query.all()
[]
>>> UserFactory() # Creates a new user
<User: john>
>>> User.query.all()
[<User: john>]
>>> UserFactory() # Fetches the existing user
<User: john>
>>> User.query.all() # No new user!
[<User: john>]
>>> UserFactory(username='jack') # Creates another user
<User: jack>
>>> User.query.all()
[<User: john>, <User: jack>]
请考虑到,当使用
sqlalchemy_get_or_create
时,传递给工厂的任何新值都不用于更新现有模型。