"""Provide Router for db migrations.""" import os from contextlib import suppress from peewee_migrate import Router as PeeweeRouter from peewee_migrate.router import void __all__ = ["Router"] class Router(PeeweeRouter): """Like peewee_migrate.Router but supports multiple migrations dirs.""" # this is a slightly edited version from peewee_migrate.router.Router def __init__(self, database, migrations_dirs, **kwargs): super().__init__(database, migrate_dir=migrations_dirs[0], **kwargs) self.migrations_dirs = migrations_dirs @property def todo(self): """Scan migrations in file system.""" for migrate_dir in self.migrations_dirs: if not os.path.exists(migrate_dir): self.logger.warn( "Migration directory: %s does not exist.", migrate_dir ) os.makedirs(migrate_dir) migration_names = [] for migrate_dir in self.migrations_dirs: migration_names += sorted( f[: -len(".py")] for f in os.listdir(migrate_dir) if self.filemask.match(f) ) return migration_names def read(self, name): """Read migration from file.""" scope = {} for migrate_dir in self.migrations_dirs: with suppress(FileNotFoundError): with open(os.path.join(migrate_dir, name + ".py")) as f: code = compile( f.read(), "", "exec", dont_inherit=True ) exec(code, scope) return scope.get("migrate", void), scope.get("rollback", void)