Flask-Migrate

dependency_manager.py

import os
import platform

__author__ = "Ronivaldo Sampaio"
__copyright__ = "Kontus (www.kontus.com.br)"
__credits__ = ["Ronivaldo Sampaio"]
__version__ = "1.0.0"
__maintainer__ = "Ronivaldo Sampaio"
__email__ = "ronivaldo@gmail.com"
__status__ = "Production"

def install_dependency(dependency):
    if platform.system() == 'Linux':
        command_prefix = 'sudo '
        command_sufix = ' > /dev/null'
    else:
        command_prefix = ''
        command_sufix = ' > nul 2>&1'
    os.system('{}pip install {}{}'.format(command_prefix, dependency, command_sufix))

try:
	from flask_migrate import Migrate, upgrade
except ImportError:
	install_dependency('Flask-Migrate==2.5.2')   

how_to.md

Install Flask-Migrate

sudo pip install Flask-Migrate==2.5.2

First Migrate actions

1. enable Flask to the app:

set FLASK_APP=service_monitor_rest.py

2. Add in the distance_rest.py:

from flask_migrate import Migrate, upgrade

db = SQLAlchemy(app)
migrate = Migrate(app, db)

	#db.create_all()
    with app.app_context():
        upgrade()

3. Create files and folders:

flask db init

4. Make sure the server is not running.

5. Rename or delete the database to force the creation of the DDL script.

6. Create the first migration:

flask db migrate --message "initial_tables"

7. Add in the very first version of migration the check for existing tables:

from sqlalchemy.engine.reflection import Inspector

...
def upgrade():
    conn = op.get_bind()
    inspector = Inspector.from_engine(conn)
    tables = inspector.get_table_names()

    if 'monitor_server' not in tables:
        op.create_table('monitor_server',
...

8. Test the server

python service_monitor_rest.py

Migrate to another DB version

1. Add in the [service]_rest.py the new Models Definitions:

class SettingService(db.Model):
    id=db.Column(db.Integer, primary_key=True, autoincrement=False)
    warehouse_id=db.Column(db.Integer, nullable=False)
    target=db.Column(db.String(50), nullable=False)
    setting_type=db.Column(db.String(50), nullable=False)
    ST_NOT_PROCESSED = 1
    ST_PROCESSED = 2
    processed=db.Column(db.Integer, nullable=False, default=ST_NOT_PROCESSED)
    setting_text=db.Column(db.Text, nullable=False)

    def is_processed(self):
        return self.processed == self.ST_PROCESSED

    def json_to_setting_text(self, json_value):
        if json_value is not None:
            self.setting_text = json.dumps(json_value)

    def setting_text_to_json(self):
        if self.setting_text is not None:
            return json.loads(self.setting_text)
        return {}

2. Make sure the server is not running.

3. Keep the database in the previous version, so Migrate will create the new upgrade DDL script.

4. Create the new migration:

flask db migrate --message "setting_service_table"

5. Test migration

python service_monitor_rest.py

6. Optionally, add in the very first version of migration the check for existing tables:

from sqlalchemy.engine.reflection import Inspector

...
def upgrade():
    conn = op.get_bind()
    inspector = Inspector.from_engine(conn)
    tables = inspector.get_table_names()

    if 'setting_service' not in tables:
        op.create_table('setting_service',
...

service_rest.py

#!/usr/bin/env python
from __future__ import print_function
import dependency_manager