如何在运行时添加ManyToManyField？

Question

这就是我想要做的事：

def test_basic_addition(self): 
    # create field 
    f = models.ManyToManyField(to=X, related_name='bar') 
    f.contribute_to_class(Y, 'x') 

    # create table 
    field = Y._meta.get_field_by_name('x')[0] 
    through = field.rel.through 

    fields = tuple((field.name, field) for field in through._meta.fields) 

    db.create_table(through._meta.db_table, fields) 
    db.create_unique(through._meta.db_table, 
     ['%s_id' % name for name, f in fields 
      if isinstance(f, models.ForeignKey)]) 


    x = X(name='foo') 
    x.save() 
    y = Y() 
    y.save() 
    y.x.add(x) 

    print y.x.all() 

引发的异常是：

E 
====================================================================== 
ERROR: test_basic_addition (test_app.tests.SimpleTest) 
---------------------------------------------------------------------- 
Traceback (most recent call last): 
    File "/home/jpic/test_project/test_app/tests.py", line 41, in test_basic_addition 
    print y.x.all() 
    File "/home/jpic/env/local/lib/python2.7/site-packages/django/db/models/manager.py", line 116, in all 
    return self.get_query_set() 
    File "/home/jpic/env/local/lib/python2.7/site-packages/django/db/models/fields/related.py", line 543, in get_query_set 
    return super(ManyRelatedManager, self).get_query_set().using(db)._next_is_sticky().filter(**self.core_filters) 
    File "/home/jpic/env/local/lib/python2.7/site-packages/django/db/models/query.py", line 621, in filter 
    return self._filter_or_exclude(False, *args, **kwargs) 
    File "/home/jpic/env/local/lib/python2.7/site-packages/django/db/models/query.py", line 639, in _filter_or_exclude 
    clone.query.add_q(Q(*args, **kwargs)) 
    File "/home/jpic/env/local/lib/python2.7/site-packages/django/db/models/sql/query.py", line 1250, in add_q 
    can_reuse=used_aliases, force_having=force_having) 
    File "/home/jpic/env/local/lib/python2.7/site-packages/django/db/models/sql/query.py", line 1122, in add_filter 
    process_extras=process_extras) 
    File "/home/jpic/env/local/lib/python2.7/site-packages/django/db/models/sql/query.py", line 1316, in setup_joins 
    "Choices are: %s" % (name, ", ".join(names))) 
FieldError: Cannot resolve keyword 'bar' into field. Choices are: id, name, users 

---------------------------------------------------------------------- 
Ran 1 test in 0.017s 

FAILED (errors=1) 
Destroying test database for alias 'default'... 

del X._meta._name_map不会做的伎俩，但我想这是正常的，因为它是一个反向字段来从另一个型号，Y.

无论如何，您可以结帐我的test_project它隔离我的问题，然后在您的工作副本中运行./manage.py test test_app重现此问题。

Answer 1

解决方案是实施疯狂的黑客Django的AppCache，显然is a pain。

最简洁的解决方案是使用SQLAlchemy，这里的概念的一个可怕的证明，重构和一些锁定功能将成为一个Python包后：

import unittest 

import sqlalchemy as sa 
from sqlalchemy import orm 
from sqlalchemy.ext.declarative import declarative_base 

from alembic.operations import Operations 
from alembic.migration import MigrationContext 

Base = declarative_base() 
engine = sa.create_engine('mysql://root:root@localhost/jpic') 
Session = orm.sessionmaker(bind=engine) 
session = Session() 

conn = engine.connect() 
ctx = MigrationContext.configure(conn) 
op = Operations(ctx) 

for table in ('person_car', 'cars', 'houses', 'persons'): 
    op.drop_table(table) 


class PersonTest(unittest.TestCase): 
    def test_000_create_table(self): 

     self.__class__.Person = type('Person', (Base,), {'__tablename__': 'persons', 
      'id': sa.Column(sa.Integer, primary_key=True)}) 

     self.__class__.Car = type('Car', (Base,), {'__tablename__': 'cars', 
      'id': sa.Column(sa.Integer, primary_key=True)}) 

     self.__class__.House = type('House', (Base,), {'__tablename__': 'houses', 
      'id': sa.Column(sa.Integer, primary_key=True)}) 

     Base.metadata.create_all(engine) 

    def test_001_create_unicode_field(self): 
     # create the column in the table - does not add it in the class 
     field = sa.Column('unicode_field', sa.Unicode(50)) 
     op.add_column('persons', field) 

     # create the column in the class - was not done above 
     # a new instance to avoid conflicts 
     field = sa.Column('unicode_field', sa.Unicode) 
     self.__class__.Person.unicode_field = field 

     subject = self.__class__.Person(unicode_field='hello unicode field') 
     session.add(subject) 

     subject = session.query(self.__class__.Person).first() 
     self.assertEqual(subject.unicode_field, 'hello unicode field') 

    def test_002_create_foreign_key(self): 
     field = sa.Column('owner_id', sa.Integer, sa.ForeignKey('persons.id')) 
     op.add_column('houses', field) 

     # create fk 
     op.create_foreign_key('fk_house_owner', 'houses', 'persons', ['owner_id'], ['id']) 

     field = sa.Column('owner_id', sa.Integer, sa.ForeignKey('persons.id')) 
     relation = orm.relationship('Person', 
       backref=orm.backref('houses', lazy='dynamic')) 
     self.__class__.House.owner_id = field 
     self.__class__.House.owner = relation 


     owner = session.query(self.__class__.Person).first() 

     house = self.__class__.House() 
     house.owner = owner 
     session.add(house) 

     house = session.query(self.__class__.House).first() 

     self.assertEqual(house.owner, owner) 
     # also test the reverse relation 
     self.assertEqual(owner.houses.all(), [house]) 

    def test_003_create_many_to_many(self): 
     association_table = sa.Table('person_car', Base.metadata, 
      sa.Column('person_id', sa.Integer, sa.ForeignKey('persons.id')), 
      sa.Column('car_id', sa.Integer, sa.ForeignKey('cars.id')) 
     ) 

     Base.metadata.create_all(engine) 

     self.__class__.Person.cars = orm.relationship('Car', 
       secondary=association_table, 
       backref=orm.backref('persons', lazy='dynamic')) 

     user1 = session.query(self.__class__.Person).first() 
     user2 = self.__class__.Person() 
     session.add(user2) 

     car1 = self.__class__.Car() 
     session.add(car1) 
     car2 = self.__class__.Car() 
     session.add(car2) 

     user1.cars.append(car1) 
     car2.persons.append(user1) 
     session.commit() 

     fresh_user1 = session.query(self.__class__.Person).get(user1.id) 
     self.assertEqual(len(fresh_user1.cars), 2) 
     self.assertTrue(car1 in fresh_user1.cars) 
     self.assertTrue(car2 in fresh_user1.cars) 

     fresh_car1 = session.query(self.__class__.Car).get(car1.id) 
     self.assertEqual(fresh_car1.persons.all(), [user1]) 

注意，该工程对MySQL的，但不是SQLite的，也不Postgres的。

Hello Flask and SQLAlchemy B）