libs/dmigrations/migration_db.py

   1 from migration_loader import load_migration_from_path
   2 from exceptions import *
   3
   4 import os, sys
   5 import re
   6
   7 class MigrationDb(object):
   8
   9     def __init__(self, directory=None, migrations=None):
  10         self.directory = directory
  11         self._migrations = migrations
  12
  13     @property
  14     def migrations(self):
  15         """
  16         Lazy migrations property. Mostly to make overriding in tests easier.
  17         """
  18         if self._migrations == None:
  19             self.populate_migrations_from_directory()
  20
  21         return self._migrations
  22
  23     def populate_migrations_from_ls(self, ls):
  24         """
  25         Populate a list of migrations based on directory listing.
  26         Separate from populate_migrations_from_directory for easier testing.
  27         """
  28         self._migrations = [
  29             re.sub(r'\.py$', '', file_name)
  30             for file_name in ls
  31             if re.search(r'^\d+_.*\.py$', file_name)
  32         ]
  33         self.warn_if_duplicate_migration_numbers()
  34
  35     def migration_number(self, migration):
  36         """
  37         Return migration number based on migration name like <int>_<anything>
  38         """
  39         m = re.search(r'(\d+)_', migration)
  40         if m:
  41             return int(m.group(1))
  42         else:
  43             raise Exception(u"%s is not a valid migration name" % migration)
  44
  45     def migration_sort_key(self, migration):
  46         """
  47         Return sort key for migrations - by number first, then by ascii.
  48         """
  49         return (self.migration_number(migration), migration)
  50
  51     def sort_migrations(self, migration_list):
  52         """
  53         Return sorted list of migrations.
  54         """
  55         return sorted(migration_list, key=self.migration_sort_key)
  56
  57     def warn_if_duplicate_migration_numbers(self):
  58         """
  59         Warn if there are multiple migrations with the same number.
  60         This situation is explicitly SUPPORTED, so it's not an error,
  61         but more likely than not it's not what you want to do.
  62         """
  63         by_number = {}
  64
  65         for migration_name in self.migrations:
  66             i = self.migration_number(migration_name)
  67             by_number.setdefault(i, [])
  68             by_number[i].append(migration_name)
  69
  70         for number in sorted(by_number.keys()):
  71             if len(by_number[number]) > 1:
  72                 self.warn(
  73                     u"There are multiple migrations with the same number "
  74                      "%d: %s" % (number, ", ".join(
  75                         sorted(by_number[number])
  76                     ))
  77                 )
  78
  79     def warn(self, warning):
  80         """
  81         Print warning.
  82         Separate function for easy testing.
  83         """
  84         print >>sys.stderr, u"Warning: %s" % warning
  85
  86     def populate_migrations_from_directory(self):
  87         """
  88         Populate list of migrations based on self.directory.
  89         """
  90         if self.directory == None:
  91             self.populate_migrations_from_ls([])
  92         else:
  93             self.populate_migrations_from_ls(os.listdir(self.directory))
  94
  95     def list(self):
  96         """
  97         Return ordered list of migrations in the database.
  98         """
  99         return self.sort_migrations(self.migrations)
 100
 101     def is_dev_migration(self, name):
 102         """
 103         Migration is a DEV migration if it has string "_DEV_" in its name.
 104         """
 105         return bool(re.search(r'_DEV_', name))
 106
 107     def find_unique_migration_by_number(self, number):
 108         matching_migrations = [
 109             name for name in self.migrations
 110             if self.migration_number(name) == number
 111         ]
 112
 113         if len(matching_migrations) == 0:
 114             return None
 115         elif len(matching_migrations) == 1:
 116             return matching_migrations[0]
 117         else:
 118             raise AmbiguousMigrationNameError(number)
 119
 120     def force_resolve_migration_name(self, name):
 121         """
 122         Take either full name or a number, and return full name for an
 123         existing migration.
 124         """
 125         if name in self.migrations:
 126             return name
 127         elif re.search(r'^\d+$', str(name)):
 128             resolved_migration_name = self.find_unique_migration_by_number(
 129                 int(name)
 130             )
 131             if resolved_migration_name is None:
 132                 raise NoSuchMigrationError(name)
 133             else:
 134                 return resolved_migration_name
 135         else:
 136             raise NoSuchMigrationError(name)
 137
 138     def resolve_migration_path(self, name):
 139         """
 140         Return path for existing migration name.
 141         """
 142         name = self.force_resolve_migration_name(name)
 143         return os.path.join(self.directory, name + ".py")
 144
 145     def migration_path(self, name):
 146         """
 147         Return path for new migration name.
 148         """
 149         number = 1 + max([0] + [
 150             self.migration_number(migration) for migration in self.migrations
 151         ])
 152
 153         return u"%s/%03d_%s.py" % (self.directory, number, name)
 154
 155     def load_migration_object(self, name):
 156         """
 157         Get migration with given name or number.
 158         """
 159
 160         name = self.force_resolve_migration_name(name)
 161         full_path = self.resolve_migration_path(name)
 162         dev = self.is_dev_migration(name)
 163
 164         return load_migration_from_path(full_path, dev=dev)