Привет, делаю миграции в peewee по статье с хабра -
https://habr.com/ru/post/262697/
Есть абстрактный класс мигратор
file: ./migrator.py
class Migrator(object):
"""
Migration interface
"""
__metaclass__ = abc.ABCMeta
db_connection = PostgresqlDatabase(config.DB_NAME,
user=config.USERNAME,
password=config.PASSWORD,
host=config.HOST,
port=config.PORT,
autorollback=True)
# db_connection is a Borg instance
connection = db_connection.connection
migrator = PostgresqlMigrator(db_connection.connection)
@abc.abstractproperty
def migrations(self):
"""
List of the migrations dictionaries
:param self: class instance
:return: list
"""
return [
{'statement': 1 != 2, 'migration': ['list', 'of', 'migration', 'options'],
'migration_kwargs': {}, 'pre_migrations': list(), 'post_migrations': list()}
] # Just an example
def migrate(self):
"""
Run migrations
"""
for migration in self.migrations:
if migration['statement']:
# Run scripts before the migration
pre_migrations = migration.get('pre_migrations', list())
for pre_m in pre_migrations:
pre_m()
# Migrate
with self.db_connection.connection.transaction():
migration_kwargs = migration.get('migration_kwargs', {})
migrate(*migration['migration'], **migration_kwargs)
# Run scripts after the migration
post_migrations = migration.get('post_migrations', list())
for post_m in post_migrations:
post_m()
Также в этом файле есть функции которые ищут модули в папке ./migrations/migrations
file: ./migrator.py
def get_migration_modules(packages=None):
"""
Get python modules with migrations
:param packages: iterable - list or tuple with packages names for the searching
:return: list - ('module.path', 'module_name')
"""
if packages is None:
packages = []
# List of the modules to migrate
migration_modules = []
for pack in packages:
migration_module = __import__(
pack, globals(), locals(), fromlist=['migrations'])
with contextlib.suppress(AttributeError):
# Check, that imported object is module
if inspect.ismodule(migration_module.migrations):
# Find submodules inside the module
migration_modules.extend((migration_module.migrations.__name__, modname) for importer, modname, ispkg in pkgutil.iter_modules(
migration_module.migrations.__path__) if re.match(r'^\d{3,}_migration_[\d\w_]+$', modname) and not ispkg)
# Unregister module
sys.modules.pop(migration_module.__name__)
return migration_modules
def get_migration_classes(migration_modules):
"""
Get list of the migration classes
:type migration_modules: iterable
:param migration_modules: array with a migration modules
:return: list
"""
migration_classes = []
for mig_mod, m in migration_modules:
mig = __import__(mig_mod, globals(), locals(), fromlist=[m])
with contextlib.suppress(AttributeError):
target_module = mig.__getattribute__(m)
# Check, that imported object is module
if inspect.ismodule(target_module):
migration_classes.extend(obj for name, obj in inspect.getmembers(target_module) if inspect.isclass(obj) and issubclass(obj, Migrator) and obj != Migrator)
# Remove imported module from the stack
sys.modules.pop(mig.__name__)
return migration_classes
И также запуск поиска и авто запуска миграций
file: ./migrator.py
if __name__ == '__main__':
# Get modules with migrations
m_mods = get_migration_modules(packages=['migrations'])
# Get migration classes
m_classes = get_migration_classes(m_mods)
print(m_classes)
# Execute migrations
for m_class in m_classes:
mig = m_class()
mig.migrate()
Пока что в папке
./migrations/migrations лежит одна миграция
001_migration_add_new_field.py
Вот она:
file: ./migrations/migrations/001_migration_add_new_field.py
from migrator import Migrator
from utils.database import db as mig_db
class AddFieldInPoll(Migrator):
table_name = mig_db.Poll._meta.table_name
def __field_not_exists(self):
"""
Check, that new field does not exists
:return: bool
"""
q = 'SELECT 1 from information_schema.columns where column_name = \'group_for_send\' and table_name = \'{0}\''.format(
self.table_name)
cursor = self.connection.execute_sql(q)
result = int(cursor.fetchone()[0])
return result == 0
@property
def migrations(self):
print('001_migration')
return [
# add group_for_send column
{
'statement': self.__field_not_exists(),
'migration': [self.migrator.add_column(self.table_name, 'group_for_send', mig_db.Poll.group_for_send)],
}
]
Но миграция не находится так как в этом моменте не очень понятно работает issubclass
Момент:
file: ./migrator.py
migration_classes.extend(obj for name, obj in inspect.getmembers(target_module) if inspect.isclass(obj) and issubclass(obj, Migrator) and obj != Migrator)
В этом цикле в самой первой итерации
obj = <class 'migrations.migrations.001_migration_add_new_field.AddFieldInPoll'>
Первое условие выдаёт True: inspect.isclass(obj)
Третье выдаёт True: obj != Migrator
А вот второе выдаёт False: issubclass(obj, Migrator)
Хотя родитель класса AddFieldInPoll это класс Migrator. Почему так и как это исправить?