Merge pull request #6273 from freqtrade/pg_migrations

Update database migrations for PG and mariadb

freqtrade/persistence/migrations.py

@@ -28,7 +28,36 @@ def get_backup_name(tabs, backup_prefix: str):
     return table_back_name
 
 
-def migrate_trades_table(decl_base, inspector, engine, table_back_name: str, cols: List):
+def get_last_sequence_ids(engine, trade_back_name, order_back_name):
+    order_id: int = None
+    trade_id: int = None
+
+    if engine.name == 'postgresql':
+        with engine.begin() as connection:
+            trade_id = connection.execute(text("select nextval('trades_id_seq')")).fetchone()[0]
+            order_id = connection.execute(text("select nextval('orders_id_seq')")).fetchone()[0]
+        with engine.begin() as connection:
+            connection.execute(text(
+                f"ALTER SEQUENCE orders_id_seq rename to {order_back_name}_id_seq_bak"))
+            connection.execute(text(
+                f"ALTER SEQUENCE trades_id_seq rename to {trade_back_name}_id_seq_bak"))
+    return order_id, trade_id
+
+
+def set_sequence_ids(engine, order_id, trade_id):
+
+    if engine.name == 'postgresql':
+        with engine.begin() as connection:
+            if order_id:
+                connection.execute(text(f"ALTER SEQUENCE orders_id_seq RESTART WITH {order_id}"))
+            if trade_id:
+                connection.execute(text(f"ALTER SEQUENCE trades_id_seq RESTART WITH {trade_id}"))
+
+
+def migrate_trades_and_orders_table(
+        decl_base, inspector, engine,
+        trade_back_name: str, cols: List,
+        order_back_name: str):
     fee_open = get_column_def(cols, 'fee_open', 'fee')
     fee_open_cost = get_column_def(cols, 'fee_open_cost', 'null')
     fee_open_currency = get_column_def(cols, 'fee_open_currency', 'null')
@@ -64,11 +93,20 @@ def migrate_trades_table(decl_base, inspector, engine, table_back_name: str, col
 
     # Schema migration necessary
     with engine.begin() as connection:
-        connection.execute(text(f"alter table trades rename to {table_back_name}"))
+        connection.execute(text(f"alter table trades rename to {trade_back_name}"))
+
     with engine.begin() as connection:
         # drop indexes on backup table in new session
-        for index in inspector.get_indexes(table_back_name):
+        for index in inspector.get_indexes(trade_back_name):
+            if engine.name == 'mysql':
+                connection.execute(text(f"drop index {index['name']} on {trade_back_name}"))
+            else:
                 connection.execute(text(f"drop index {index['name']}"))
+
+    order_id, trade_id = get_last_sequence_ids(engine, trade_back_name, order_back_name)
+
+    drop_orders_table(engine, order_back_name)
+
     # let SQLAlchemy create the schema as required
     decl_base.metadata.create_all(engine)
 
@@ -100,9 +138,12 @@ def migrate_trades_table(decl_base, inspector, engine, table_back_name: str, col
             {sell_order_status} sell_order_status,
             {strategy} strategy, {buy_tag} buy_tag, {timeframe} timeframe,
             {open_trade_value} open_trade_value, {close_profit_abs} close_profit_abs
-            from {table_back_name}
+            from {trade_back_name}
             """))
 
+    migrate_orders_table(engine, order_back_name, cols)
+    set_sequence_ids(engine, order_id, trade_id)
+
 
 def migrate_open_orders_to_trades(engine):
     with engine.begin() as connection:
@@ -121,19 +162,18 @@ def migrate_open_orders_to_trades(engine):
         """))
 
 
-def migrate_orders_table(decl_base, inspector, engine, table_back_name: str, cols: List):
+def drop_orders_table(engine, table_back_name: str):
-    # Schema migration necessary
+    # Drop and recreate orders table as backup
+    # This drops foreign keys, too.
 
     with engine.begin() as connection:
-        connection.execute(text(f"alter table orders rename to {table_back_name}"))
+        connection.execute(text(f"create table {table_back_name} as select * from orders"))
+        connection.execute(text("drop table orders"))
 
-    with engine.begin() as connection:
+
-        # drop indexes on backup table in new session
+def migrate_orders_table(engine, table_back_name: str, cols: List):
-        for index in inspector.get_indexes(table_back_name):
-            connection.execute(text(f"drop index {index['name']}"))
 
     # let SQLAlchemy create the schema as required
-    decl_base.metadata.create_all(engine)
     with engine.begin() as connection:
         connection.execute(text(f"""
             insert into orders ( id, ft_trade_id, ft_order_side, ft_pair, ft_is_open, order_id,
@@ -155,11 +195,16 @@ def check_migrate(engine, decl_base, previous_tables) -> None:
     cols = inspector.get_columns('trades')
     tabs = get_table_names_for_table(inspector, 'trades')
     table_back_name = get_backup_name(tabs, 'trades_bak')
+    order_tabs = get_table_names_for_table(inspector, 'orders')
+    order_table_bak_name = get_backup_name(order_tabs, 'orders_bak')
 
-    # Check for latest column
+    # Check if migration necessary
+    # Migrates both trades and orders table!
     if not has_column(cols, 'buy_tag'):
-        logger.info(f'Running database migration for trades - backup: {table_back_name}')
+        logger.info(f"Running database migration for trades - "
-        migrate_trades_table(decl_base, inspector, engine, table_back_name, cols)
+                    f"backup: {table_back_name}, {order_table_bak_name}")
+        migrate_trades_and_orders_table(
+            decl_base, inspector, engine, table_back_name, cols, order_table_bak_name)
         # Reread columns - the above recreated the table!
         inspector = inspect(engine)
         cols = inspector.get_columns('trades')
@@ -167,12 +212,3 @@ def check_migrate(engine, decl_base, previous_tables) -> None:
     if 'orders' not in previous_tables and 'trades' in previous_tables:
         logger.info('Moving open orders to Orders table.')
         migrate_open_orders_to_trades(engine)
-    else:
-        cols_order = inspector.get_columns('orders')
-
-        if not has_column(cols_order, 'average'):
-            tabs = get_table_names_for_table(inspector, 'orders')
-            # Empty for now - as there is only one iteration of the orders table so far.
-            table_back_name = get_backup_name(tabs, 'orders_bak')
-
-            migrate_orders_table(decl_base, inspector, engine, table_back_name, cols)

tests/test_persistence.py

@@ -8,11 +8,12 @@ from unittest.mock import MagicMock
 
 import arrow
 import pytest
-from sqlalchemy import create_engine, inspect, text
+from sqlalchemy import create_engine, text
 
 from freqtrade import constants
 from freqtrade.exceptions import DependencyException, OperationalException
 from freqtrade.persistence import LocalTrade, Order, Trade, clean_dry_run_db, init_db
+from freqtrade.persistence.migrations import get_last_sequence_ids, set_sequence_ids
 from tests.conftest import create_mock_trades, create_mock_trades_usdt, log_has, log_has_re
 
 
@@ -600,7 +601,8 @@ def test_migrate_new(mocker, default_conf, fee, caplog):
     assert trade.stoploss_last_update is None
     assert log_has("trying trades_bak1", caplog)
     assert log_has("trying trades_bak2", caplog)
-    assert log_has("Running database migration for trades - backup: trades_bak2", caplog)
+    assert log_has("Running database migration for trades - backup: trades_bak2, orders_bak0",
+                   caplog)
     assert trade.open_trade_value == trade._calc_open_trade_value()
     assert trade.close_profit_abs is None
 
@@ -613,65 +615,6 @@ def test_migrate_new(mocker, default_conf, fee, caplog):
     assert orders[1].order_id == 'stop_order_id222'
     assert orders[1].ft_order_side == 'stoploss'
 
-    caplog.clear()
-    # Drop latest column
-    with engine.begin() as connection:
-        connection.execute(text("alter table orders rename to orders_bak"))
-    inspector = inspect(engine)
-
-    with engine.begin() as connection:
-        for index in inspector.get_indexes('orders_bak'):
-            connection.execute(text(f"drop index {index['name']}"))
-        # Recreate table
-        connection.execute(text("""
-            CREATE TABLE orders (
-                id INTEGER NOT NULL,
-                ft_trade_id INTEGER,
-                ft_order_side VARCHAR NOT NULL,
-                ft_pair VARCHAR NOT NULL,
-                ft_is_open BOOLEAN NOT NULL,
-                order_id VARCHAR NOT NULL,
-                status VARCHAR,
-                symbol VARCHAR,
-                order_type VARCHAR,
-                side VARCHAR,
-                price FLOAT,
-                amount FLOAT,
-                filled FLOAT,
-                remaining FLOAT,
-                cost FLOAT,
-                order_date DATETIME,
-                order_filled_date DATETIME,
-                order_update_date DATETIME,
-                PRIMARY KEY (id),
-                CONSTRAINT _order_pair_order_id UNIQUE (ft_pair, order_id),
-                FOREIGN KEY(ft_trade_id) REFERENCES trades (id)
-            )
-            """))
-
-        connection.execute(text("""
-        insert into orders ( id, ft_trade_id, ft_order_side, ft_pair, ft_is_open, order_id, status,
-            symbol, order_type, side, price, amount, filled, remaining, cost, order_date,
-            order_filled_date, order_update_date)
-            select id, ft_trade_id, ft_order_side, ft_pair, ft_is_open, order_id, status,
-            symbol, order_type, side, price, amount, filled, remaining, cost, order_date,
-            order_filled_date, order_update_date
-            from orders_bak
-        """))
-
-    # Run init to test migration
-    init_db(default_conf['db_url'], default_conf['dry_run'])
-
-    assert log_has("trying orders_bak1", caplog)
-
-    orders = Order.query.all()
-    assert len(orders) == 2
-    assert orders[0].order_id == 'buy_order'
-    assert orders[0].ft_order_side == 'buy'
-
-    assert orders[1].order_id == 'stop_order_id222'
-    assert orders[1].ft_order_side == 'stoploss'
-
 
 def test_migrate_mid_state(mocker, default_conf, fee, caplog):
     """
@@ -733,7 +676,40 @@ def test_migrate_mid_state(mocker, default_conf, fee, caplog):
     assert trade.initial_stop_loss == 0.0
     assert trade.open_trade_value == trade._calc_open_trade_value()
     assert log_has("trying trades_bak0", caplog)
-    assert log_has("Running database migration for trades - backup: trades_bak0", caplog)
+    assert log_has("Running database migration for trades - backup: trades_bak0, orders_bak0",
+                   caplog)
+
+
+def test_migrate_get_last_sequence_ids():
+    engine = MagicMock()
+    engine.begin = MagicMock()
+    engine.name = 'postgresql'
+    get_last_sequence_ids(engine, 'trades_bak', 'orders_bak')
+
+    assert engine.begin.call_count == 2
+    engine.reset_mock()
+    engine.begin.reset_mock()
+
+    engine.name = 'somethingelse'
+    get_last_sequence_ids(engine, 'trades_bak', 'orders_bak')
+
+    assert engine.begin.call_count == 0
+
+
+def test_migrate_set_sequence_ids():
+    engine = MagicMock()
+    engine.begin = MagicMock()
+    engine.name = 'postgresql'
+    set_sequence_ids(engine, 22, 55)
+
+    assert engine.begin.call_count == 1
+    engine.reset_mock()
+    engine.begin.reset_mock()
+
+    engine.name = 'somethingelse'
+    set_sequence_ids(engine, 22, 55)
+
+    assert engine.begin.call_count == 0
 
 
 def test_adjust_stop_loss(fee):