|
# -*- coding: utf-8 -*- |
|
import datetime |
|
from south.db import db |
|
from south.v2 import DataMigration |
|
from django.contrib.contenttypes import generic |
|
from django.contrib.contenttypes.models import ContentType |
|
from django.db import models |
|
|
|
class Migration(DataMigration): |
|
|
|
def content_type_for_model(self, orm, model): |
|
""" |
|
Get a ContentType for the given model |
|
""" |
|
# Can't just use ContentType.objects.get_for_model |
|
return orm['contenttypes.ContentType'].objects.get( |
|
app_label=model._meta.app_label, |
|
model=model._meta.model_name, |
|
) |
|
|
|
def content_types(self, orm): |
|
""" |
|
Model to ContentType mappings for all "interesting" models |
|
""" |
|
return dict( |
|
(model, self.content_type_for_model(orm, model).pk) |
|
for model |
|
in (orm.Entity, # List of entities; only one for the example |
|
) |
|
) |
|
|
|
def forwards(self, orm): |
|
""" |
|
Add duplicate data for the content types from the link that exists |
|
already in the one to one relation. |
|
""" |
|
# If there is no data, there's no point in migrating |
|
if not orm.PostalAddress.objects.exists(): |
|
return |
|
|
|
CONTENT_TYPES = self.content_types(orm) |
|
|
|
for model, content_type_id in CONTENT_TYPES.iteritems(): |
|
for obj in model.objects.all().iterator(): |
|
|
|
postal_address = obj.postal_address |
|
if postal_address: |
|
# Unfortunately, we can't just use: |
|
# postal_address.content_object = obj |
|
# |
|
# See http://south.readthedocs.org/en/latest/generics.html |
|
# And http://south.readthedocs.org/en/0.8.4/generics.html |
|
postal_address.content_type_id = content_type_id |
|
postal_address.object_id = obj.pk |
|
|
|
postal_address.save(force_update=True) |
|
|
|
invalid_count = orm.PostalAddress.objects.filter(content_type_id=1, |
|
object_id=1, |
|
).count() |
|
assert invalid_count == 0, ("No PostalAddress objects still exist " |
|
"with default values") |
|
|
|
def backwards(self, orm): |
|
""" |
|
Add duplicate data for the one to one from the link that exists already |
|
in the content types relation. |
|
""" |
|
# If there is no data, there's no point in migrating |
|
if not orm.PostalAddress.objects.exists(): |
|
return |
|
|
|
CONTENT_TYPES = self.content_types(orm) |
|
MODELS_LIST = CONTENT_TYPES.keys() |
|
MODEL_MAPPING = dict( |
|
(content_type_id, model) |
|
for model, content_type_id |
|
in CONTENT_TYPES.items() |
|
) |
|
|
|
# Blank all current relations |
|
for model in MODELS_LIST: |
|
model.objects.all().update(postal_address=None) |
|
|
|
# Migrate addresses |
|
for address in orm.PostalAddress.objects.all().iterator(): |
|
entity = MODEL_MAPPING[address.content_type_id].objects.get( |
|
pk=address.object_id |
|
) |
|
entity.postal_address_id = address.pk |
|
entity.save(force_update=True) |
|
|
|
# Check that the migration was successful |
|
for model in MODELS_LIST: |
|
content_type_id = CONTENT_TYPES[model] |
|
|
|
# Check every model |
|
for entity in model.objects.all().iterator(): |
|
gfr = orm.PostalAddress.objects.filter( |
|
content_type_id=content_type_id, |
|
object_id=entity.pk, |
|
) |
|
if gfr.count() == 0: |
|
assert entity.postal_address_id == None, ( |
|
"OTO postal address null where no GFR postal address") |
|
else: |
|
same = entity.postal_address_id == gfr.first().pk |
|
assert same, ("OTO postal address id same as GFR postal " |
|
"address id") |
|
|
|
models = { |
|
# ... auto generated ... |
|
} |