Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def model(self):
""" Ensure that migrations has prepared to run. """
# Initialize MigrationHistory model
MigrateHistory._meta.database = self.app.plugins.peewee.database
try:
MigrateHistory.create_table()
except pw.OperationalError:
pass
return MigrateHistory
""" Initialize the application. """
super().setup(app)
if 'manage' not in app.plugins:
raise PluginException('Peewee plugin requires Manage plugin initialized before.')
# Setup Database
self.database.initialize(connect(self.options['connection']))
self.threadpool = concurrent.futures.ThreadPoolExecutor(
max_workers=self.options['max_connections'])
if not self.options.migrations_enabled:
return
# Setup migration engine
self.router = Router(self)
self.register(MigrateHistory)
# Register migration commands
@self.app.ps.manage.command
def migrate(name:str=None):
""" Run application's migrations.
:param name: Choose a migration' name
"""
self.router.run(name)
@self.app.ps.manage.command
def create(name:str):
""" Create a migration.
:param name: Set name of migration [auto]
def model(self):
""" Ensure that migrations has prepared to run. """
# Initialize MigrationHistory model
MigrateHistory._meta.database = self.app.plugins.peewee.database
try:
MigrateHistory.create_table()
except pw.OperationalError:
pass
return MigrateHistory
def model(self):
""" Ensure that migrations has prepared to run. """
# Initialize MigrationHistory model
MigrateHistory._meta.database = self.app.plugins.peewee.database
try:
MigrateHistory.create_table()
except pw.OperationalError:
pass
return MigrateHistory