Server IP : 66.29.132.122 / Your IP : 3.144.15.18 Web Server : LiteSpeed System : Linux business142.web-hosting.com 4.18.0-553.lve.el8.x86_64 #1 SMP Mon May 27 15:27:34 UTC 2024 x86_64 User : admazpex ( 531) PHP Version : 7.2.34 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/proc/self/root/proc/thread-self/root/proc/thread-self/root/proc/self/root/proc/thread-self/root/proc/self/root/proc/self/root/proc/self/root/proc/self/root/opt/imunify360/venv/lib64/python3.11/site-packages/defence360agent/ |
Upload File : |
"""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(), "<string>", "exec", dont_inherit=True ) exec(code, scope) return scope.get("migrate", void), scope.get("rollback", void)