OLD | NEW |
(Empty) | |
| 1 from __future__ import unicode_literals |
| 2 |
| 3 from django.apps.registry import Apps |
| 4 from django.db import models |
| 5 from django.db.utils import DatabaseError |
| 6 from django.utils.encoding import python_2_unicode_compatible |
| 7 from django.utils.timezone import now |
| 8 |
| 9 from .exceptions import MigrationSchemaMissing |
| 10 |
| 11 |
| 12 class MigrationRecorder(object): |
| 13 """ |
| 14 Deals with storing migration records in the database. |
| 15 |
| 16 Because this table is actually itself used for dealing with model |
| 17 creation, it's the one thing we can't do normally via migrations. |
| 18 We manually handle table creation/schema updating (using schema backend) |
| 19 and then have a floating model to do queries with. |
| 20 |
| 21 If a migration is unapplied its row is removed from the table. Having |
| 22 a row in the table always means a migration is applied. |
| 23 """ |
| 24 |
| 25 @python_2_unicode_compatible |
| 26 class Migration(models.Model): |
| 27 app = models.CharField(max_length=255) |
| 28 name = models.CharField(max_length=255) |
| 29 applied = models.DateTimeField(default=now) |
| 30 |
| 31 class Meta: |
| 32 apps = Apps() |
| 33 app_label = "migrations" |
| 34 db_table = "django_migrations" |
| 35 |
| 36 def __str__(self): |
| 37 return "Migration %s for %s" % (self.name, self.app) |
| 38 |
| 39 def __init__(self, connection): |
| 40 self.connection = connection |
| 41 |
| 42 @property |
| 43 def migration_qs(self): |
| 44 return self.Migration.objects.using(self.connection.alias) |
| 45 |
| 46 def ensure_schema(self): |
| 47 """ |
| 48 Ensures the table exists and has the correct schema. |
| 49 """ |
| 50 # If the table's there, that's fine - we've never changed its schema |
| 51 # in the codebase. |
| 52 if self.Migration._meta.db_table in self.connection.introspection.table_
names(self.connection.cursor()): |
| 53 return |
| 54 # Make the table |
| 55 try: |
| 56 with self.connection.schema_editor() as editor: |
| 57 editor.create_model(self.Migration) |
| 58 except DatabaseError as exc: |
| 59 raise MigrationSchemaMissing("Unable to create the django_migrations
table (%s)" % exc) |
| 60 |
| 61 def applied_migrations(self): |
| 62 """ |
| 63 Returns a set of (app, name) of applied migrations. |
| 64 """ |
| 65 self.ensure_schema() |
| 66 return set(tuple(x) for x in self.migration_qs.values_list("app", "name"
)) |
| 67 |
| 68 def record_applied(self, app, name): |
| 69 """ |
| 70 Records that a migration was applied. |
| 71 """ |
| 72 self.ensure_schema() |
| 73 self.migration_qs.create(app=app, name=name) |
| 74 |
| 75 def record_unapplied(self, app, name): |
| 76 """ |
| 77 Records that a migration was unapplied. |
| 78 """ |
| 79 self.ensure_schema() |
| 80 self.migration_qs.filter(app=app, name=name).delete() |
| 81 |
| 82 def flush(self): |
| 83 """ |
| 84 Deletes all migration records. Useful if you're testing migrations. |
| 85 """ |
| 86 self.migration_qs.all().delete() |
OLD | NEW |