"""remove pickletypes menu tickets

Revision ID: 31351fc3f3e4
Revises: 364536eb33d5
Create Date: 2017-01-06 11:00:40.901000

"""


import contextlib
import datetime
import functools
import logging
import shutil
import pickle
from collections import defaultdict

from alembic import op
from sqlalchemy import (
    inspect,
    String,
    Boolean,
    Integer,
    Text,
    DateTime,
    Table,
    MetaData,
    Column,
    PickleType,
    create_engine,
)
from sqlalchemy.sql import column, table

from agent_common.v1_0.models import (
    Payment,
    PriceLevel,
    OptionSet,
    TicketItem,
    TicketItemDiscount,
    TicketItemModifier,
    TicketDiscount,
    TicketServiceCharge,
    TicketTotals,
)

from positronics_agent.v1_0.helpers import make_cache_backup_filename

logger = logging.getLogger(__name__)


# revision identifiers, used by Alembic.
revision = '31351fc3f3e4'
down_revision = '364536eb33d5'
branch_labels = None
depends_on = None

first_agent_version = '17.5.23'


@contextlib.contextmanager
def temporary_delattr(cls, *attrs):
    filtered_attrs = [attr for attr in attrs if hasattr(cls, attr)]
    stored_attrs = {attr: getattr(cls, attr) for attr in filtered_attrs}
    for attr in filtered_attrs:
        delattr(cls, attr)
    try:
        yield
    finally:
        for attr, val in stored_attrs.items():
            setattr(cls, attr, val)


def deinstrument_classes_during_call(f):
    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        with temporary_delattr(
            PriceLevel,
            "id",
            "key",
            "menu_item_id",
            "menu_modifier_id",
            "name",
            "price_per_unit",
            "_sa_class_manager",
        ), temporary_delattr(
            OptionSet,
            "id",
            "key",
            "menu_item_id",
            "menu_modifier_id",
            "required",
            "minimum",
            "maximum",
            "_sa_class_manager",
        ), temporary_delattr(
            Payment,
            "ticket_id",
            "id",
            "type",
            "amount",
            "tip",
            "change",
            "comment",
            "last4",
            "_sa_class_manager",
        ), temporary_delattr(
            TicketDiscount,
            "ticket_id",
            "id",
            "comment",
            "name",
            "discount_id",
            "value",
            "_sa_class_manager",
        ), temporary_delattr(
            TicketItem,
            "key",
            "ticket_id",
            "id",
            "name",
            "quantity",
            "split",
            "price",
            "comment",
            "sent",
            "sent_at",
            "void",
            "menu_item_id",
            "discounts",
            "modifiers",
            "_sa_class_manager",
        ), temporary_delattr(
            TicketItemDiscount,
            "ticket_item_key",
            "id",
            "comment",
            "name",
            "value",
            "discount_id",
            "_sa_class_manager",
        ), temporary_delattr(
            TicketItemModifier,
            "key",
            "id",
            "name",
            "quantity",
            "split",
            "price",
            "comment",
            "sent",
            "sent_at",
            "menu_modifier_id",
            "modifiers",
            "_sa_class_manager",
        ), temporary_delattr(
            TicketServiceCharge,
            "ticket_id",
            "id",
            "name",
            "price",
            "comment",
            "service_charge_id",
            "_sa_class_manager",
        ), temporary_delattr(
            TicketTotals,
            "ticket_id",
            "discounts",
            "due",
            "items",
            "other_charges",
            "paid",
            "service_charges",
            "tax",
            "tips",
            "total",
            "sub",
            "_sa_class_manager",
        ):
            return f(*args, **kwargs)
    return wrapper


def get_offsets(total_size, chunk_size=100):
    return list(range(0, total_size, chunk_size))


def quote(obj):
    return "".join(("\"", str(obj), "\""))


def remove_sa_instance_state(obj):
    delattr(obj, "_sa_instance_state")
    return obj


def upgrade():
    """
    Use batch-operations to support sqlite

    > with op.batch_alter_table('table_name') as batch_op:
    >     pass
    """
    production_connection = op.get_bind()

    production_filename = production_connection.engine.url.database.replace('sqlite:///', '', 1)
    backup_filename = make_cache_backup_filename(production_filename, down_revision)
    shutil.copy(production_filename, backup_filename)
    backup_engine = create_engine('sqlite:///' + backup_filename)

    try:
        _upgrade(production_connection, backup_engine)
    except:
        logger.exception(
            "Upgrade to revision %s failed. "
            "You can restore the cache from backup file %s after downgrading.",
            revision,
            backup_filename
        )
        production_connection.close()
        backup_engine.dispose()
        raise
    else:
        logger.info('Upgrade to revision %s succeeded.', revision)
        backup_engine.dispose()


@deinstrument_classes_during_call
def _upgrade(production_connection, backup_engine):
    """
    Use batch-operations to support sqlite

    > with op.batch_alter_table('table_name') as batch_op:
    >     pass
    """
    inspector = inspect(production_connection)
    tables = set(inspector.get_table_names())

    # ALTER production database
    if 'option_sets' not in tables:
        op.create_table(
            'option_sets',
            Column('key', Integer, primary_key=True, autoincrement=True),
            Column('menu_item_id', String(32), nullable=True),
            Column('menu_modifier_id', String(32), nullable=True),
            Column('id', String(32)),
            Column('menu_modifier_group_id', String(32)),
            Column('required', Boolean),
            Column('minimum', Integer),
            Column('maximum', Integer),
            mysql_charset='utf8',
            mysql_engine='InnoDB',
        )
        with op.batch_alter_table('option_sets') as batch_op:
            batch_op.create_foreign_key(
                'fk_option_sets_menu_item',
                'menu_items',
                ['menu_item_id'],
                ['id'],
                onupdate='cascade',
                ondelete='cascade',
            )
            batch_op.create_foreign_key(
                'fk_option_sets_menu_modifier',
                'modifiers',
                ['menu_modifier_id'],
                ['id'],
                onupdate='cascade',
                ondelete='cascade',
            )
            batch_op.create_foreign_key(
                'fk_option_sets_menu_modifier_group',
                'modifier_groups',
                ['menu_modifier_group_id'],
                ['id'],
                onupdate='cascade',
                ondelete='cascade',
            )
            batch_op.create_index('idx_option_sets_menu_item_id_id', ('menu_item_id', 'id'))
            batch_op.create_index('idx_option_sets_menu_modifier_id_id', ('menu_modifier_id', 'id'))

    if 'price_levels' not in tables:
        op.create_table(
            'price_levels',
            Column('key', Integer, primary_key=True, autoincrement=True),
            Column('menu_item_id', String(32), nullable=True),
            Column('menu_modifier_id', String(32), nullable=True),
            Column('id', String(32)),
            Column('name', String(128)),
            Column('price_per_unit', Integer),
            mysql_charset='utf8',
            mysql_engine='InnoDB',
        )
        with op.batch_alter_table('price_levels') as batch_op:
            batch_op.create_foreign_key(
                'fk_price_levels_menu_item',
                'menu_items',
                ['menu_item_id'],
                ['id'],
                onupdate='cascade',
                ondelete='cascade',
            )
            batch_op.create_foreign_key(
                'fk_price_levels_menu_modifier',
                'modifiers',
                ['menu_modifier_id'],
                ['id'],
                onupdate='cascade',
                ondelete='cascade',
            )
            batch_op.create_index('idx_price_levels_menu_item_id_id', ('menu_item_id', 'id'))
            batch_op.create_index('idx_price_levels_menu_modifier_id_id', ('menu_modifier_id', 'id'))

    with op.batch_alter_table('menu_items') as batch_op:
        batch_op.drop_column('price_levels')
        batch_op.drop_column('option_sets')

    with op.batch_alter_table('modifiers') as batch_op:
        batch_op.drop_column('price_levels')
        batch_op.drop_column('option_sets')

    # Data migration
    menu_items_table = table(
        'menu_items',
        column('id', String(32)),
        column('pos_id', String(32)),
        column('name', String(128)),
        column('price_per_unit', Integer),
        column('open', Boolean),
        column('in_stock', Boolean),
        column('price_levels', PickleType),
        column('option_sets', PickleType),
        column('_stub', Boolean),
    )
    modifiers_table = table(
        'modifiers',
        column('id', String(32)),
        column('pos_id', String(32)),
        column('name', String(128)),
        column('price_per_unit', Integer),
        column('open', Boolean),
        column('in_stock', Boolean),
        column('price_levels', PickleType),
        column('option_sets', PickleType),
        column('_stub', Boolean),
    )
    price_levels_table = table(
        'price_levels',
        column('menu_item_id', String(32)),
        column('menu_modifier_id', String(32)),
        column('id', String(32)),
        column('name', String(128)),
        column('price_per_unit', Integer),
    )
    option_sets_table = table(
        'option_sets',
        column('menu_item_id', String(32)),
        column('menu_modifier_id', String(32)),
        column('id', String(32)),
        column('menu_modifier_group_id', String(32)),
        column('required', Boolean),
        column('minimum', Integer),
        column('maximum', Integer),
    )
    menu_category_menu_item_table = table(
        'menu_category_menu_item',
        column('menu_category_id', String(32)),
        column('menu_item_id', String(32)),
    )
    menu_category_menu_modifier_table = table(
        'menu_category_menu_modifier',
        column('menu_category_id', String(32)),
        column('menu_modifier_id', String(32)),
    )
    modifier_group_modifier_table = table(
        'modifier_group_modifier',
        column('modifier_group_id', String(32)),
        column('modifier_id', String(32)),
    )

    # SELECT from backup database, INSERT into production database
    for _table, foreign_key, category_association_table, mods in [
        (menu_items_table, 'menu_item_id', menu_category_menu_item_table, False),
        (modifiers_table, 'menu_modifier_id', menu_category_menu_modifier_table, True),
    ]:
        # These rows are already deleted during the alter table if foreign keys
        # are turned on (as in production).  Otherwise, you may be running this
        # script directly and foreign keys aren't turned on.
        production_connection.execute(
            " DELETE FROM {category_association_table_name} ".format(
                category_association_table_name=category_association_table.name
            )
        )
        if mods:
            production_connection.execute(
                " DELETE FROM {modifier_group_modifier_table_name} ".format(
                    modifier_group_modifier_table_name=modifier_group_modifier_table.name
                )
            )

        row_count = backup_engine.execute("SELECT COUNT(1) FROM {}".format(_table.name)).scalar()
        chunk_size = 5000
        for offset in get_offsets(row_count, chunk_size=chunk_size):
            menu_object_rows = backup_engine.execute(
                " SELECT id, price_levels, option_sets "
                " FROM {table_name} "
                " LIMIT {limit} "
                " OFFSET {offset} ".format(
                    table_name=_table.name,
                    limit=chunk_size,
                    offset=offset,
                )
            ).fetchall()
            price_levels = [
                (
                    row['id'],
                    pickle.loads(row['price_levels']) if row['price_levels'] is not None else []
                )
                for row in menu_object_rows
            ]
            option_sets = [
                (
                    row['id'],
                    pickle.loads(row['option_sets']) if row['option_sets'] is not None else []
                )
                for row in menu_object_rows
            ]
            category_association_rows = backup_engine.execute(
                " SELECT menu_category_id, {foreign_key} "
                " FROM {category_association_table_name} "
                " WHERE {foreign_key} IN ({menu_object_ids}) ".format(
                    foreign_key=foreign_key,
                    category_association_table_name=category_association_table.name,
                    menu_object_ids=", ".join(quote(row['id']) for row in menu_object_rows)
                )
            ).fetchall()
            op.bulk_insert(
                price_levels_table,
                [
                    {
                        foreign_key: foreign_key_value,
                        'id': price_level.id,
                        'name': price_level.name,
                        'price_per_unit': price_level.price_per_unit
                    }
                    for foreign_key_value, price_levels in price_levels
                    for price_level in price_levels
                ]
            )
            op.bulk_insert(
                option_sets_table,
                [
                    {
                        foreign_key: foreign_key_value,
                        'id': option_set.id,
                        'menu_modifier_group_id': option_set.menu_modifier_group_id,
                        'required': option_set.required,
                        'minimum': option_set.minimum,
                        'maximum': option_set.maximum,
                    }
                    for foreign_key_value, option_sets in option_sets
                    for option_set in option_sets
                ]
            )
            op.bulk_insert(
                category_association_table,
                [
                    {
                        'menu_category_id': row['menu_category_id'],
                        foreign_key: row[foreign_key],
                    }
                    for row in category_association_rows
                ]
            )
            if mods:
                mod_group_association_rows = backup_engine.execute(
                    " SELECT modifier_group_id, modifier_id "
                    " FROM {modifier_group_modifier_table_name} "
                    " WHERE modifier_id IN ({menu_object_ids}) ".format(
                        modifier_group_modifier_table_name=modifier_group_modifier_table.name,
                        menu_object_ids=", ".join(quote(row['id']) for row in menu_object_rows)
                    )
                ).fetchall()
                op.bulk_insert(
                    modifier_group_modifier_table,
                    [
                        {
                            'modifier_group_id': row['modifier_group_id'],
                            'modifier_id': row['modifier_id'],
                        }
                        for row in mod_group_association_rows
                    ]
                )

    for maybe_held in (
        '',  # historical tickets
        'held_'  # held tickets
    ):
        # ALTER production database
        tickets_table_name = maybe_held + 'tickets'
        totals_table_name = maybe_held + 'ticket_totals'
        payments_table_name = maybe_held + 'payments'
        discounts_table_name = maybe_held + 'ticket_discounts'
        service_charges_table_name = maybe_held + 'ticket_service_charges'
        ticket_items_table_name = maybe_held + 'ticket_items'
        ticket_item_mods_table_name = maybe_held + 'ticket_item_modifiers'
        ticket_item_mod_closures_table_name = maybe_held + 'ticket_item_modifier_closures'
        ticket_item_discounts_table_name = maybe_held + 'ticket_item_discounts'

        if totals_table_name not in tables:
            op.create_table(
                totals_table_name,
                Column('ticket_id', String(32), primary_key=True),
                Column('discounts', Integer, nullable=False),
                Column('due', Integer, nullable=False),
                Column('items', Integer, nullable=False),
                Column('other_charges', Integer, nullable=False),
                Column('paid', Integer, nullable=False),
                Column('service_charges', Integer, nullable=False),
                Column('tax', Integer, nullable=False),
                Column('tips', Integer, nullable=False),
                Column('total', Integer, nullable=False),
                Column('sub', Integer, nullable=False),
                mysql_charset='utf8',
                mysql_engine='InnoDB',
            )
            with op.batch_alter_table(totals_table_name) as batch_op:
                batch_op.create_foreign_key(
                    'fk_{}_tickets'.format(totals_table_name),
                    tickets_table_name,
                    ['ticket_id'],
                    ['id'],
                    onupdate='cascade',
                    ondelete='cascade',
                )

        if payments_table_name not in tables:
            op.create_table(
                payments_table_name,
                Column('ticket_id', String(32), primary_key=True),
                Column('id', String(32), primary_key=True),
                Column('type', String(32), nullable=False),
                Column('amount', Integer, nullable=False),
                Column('tip', Integer, nullable=False),
                Column('change', Integer),
                Column('comment', Text),
                Column('last4', String(4)),
                Column('tender_type_id', String(32)),
                mysql_charset='utf8',
                mysql_engine='InnoDB',
            )
            with op.batch_alter_table(payments_table_name) as batch_op:
                batch_op.create_foreign_key(
                    'fk_{}_tickets'.format(payments_table_name),
                    tickets_table_name,
                    ['ticket_id'],
                    ['id'],
                    onupdate='cascade',
                    ondelete='cascade',
                )

        if discounts_table_name not in tables:
            op.create_table(
                discounts_table_name,
                Column('ticket_id', String(32), primary_key=True),
                Column('id', String(32), primary_key=True),
                Column('comment', Text),
                Column('name', String(128)),
                Column('discount_id', String(32)),
                Column('value', Integer, nullable=False),
                mysql_charset='utf8',
                mysql_engine='InnoDB',
            )
            with op.batch_alter_table(discounts_table_name) as batch_op:
                batch_op.create_foreign_key(
                    'fk_{}_tickets'.format(discounts_table_name),
                    tickets_table_name,
                    ['ticket_id'],
                    ['id'],
                    onupdate='cascade',
                    ondelete='cascade',
                )

        if service_charges_table_name not in tables:
            op.create_table(
                service_charges_table_name,
                Column('ticket_id', String(32), primary_key=True),
                Column('id', String(32), primary_key=True),
                Column('name', String(128)),
                Column('price', Integer, nullable=False),
                Column('comment', Text),
                Column('service_charge_id', String(32)),
                mysql_charset='utf8',
                mysql_engine='InnoDB',
            )
            with op.batch_alter_table(service_charges_table_name) as batch_op:
                batch_op.create_foreign_key(
                    'fk_{}_tickets'.format(service_charges_table_name),
                    tickets_table_name,
                    ['ticket_id'],
                    ['id'],
                    onupdate='cascade',
                    ondelete='cascade',
                )

        if ticket_item_mods_table_name not in tables:
            op.create_table(
                ticket_item_mods_table_name,
                Column('key', Integer, primary_key=True, autoincrement=True),
                Column('id', String(32)),
                Column('name', String(128)),
                Column('quantity', Integer, nullable=False),
                Column('split', Integer),
                Column('price', Integer, nullable=False),
                Column('comment', Text),
                Column('sent', Boolean),
                Column('sent_at', DateTime),
                Column('menu_modifier_id', String(32)),
                mysql_charset='utf8',
                mysql_engine='InnoDB',
            )

        if ticket_items_table_name not in tables:
            op.create_table(
                ticket_items_table_name,
                Column('key', Integer, primary_key=True, autoincrement=True),
                Column('ticket_id', String(32)),
                Column('id', String(32)),
                Column('name', String(128)),
                Column('quantity', Integer, nullable=False),
                Column('split', Integer),
                Column('price', Integer, nullable=False),
                Column('comment', Text),
                Column('sent', Boolean),
                Column('sent_at', DateTime),
                Column('void', Boolean, nullable=False),
                Column('menu_item_id', String(32)),
                mysql_charset='utf8',
                mysql_engine='InnoDB',
            )
            with op.batch_alter_table(ticket_items_table_name) as batch_op:
                batch_op.create_foreign_key(
                    'fk_{}_tickets'.format(ticket_items_table_name),
                    tickets_table_name,
                    ['ticket_id'],
                    ['id'],
                    onupdate='cascade',
                    ondelete='cascade',
                )
                batch_op.create_index(
                    'idx_{}_id_id'.format(tickets_table_name[:-1]),
                    ('ticket_id', 'id'),
                )

        if ticket_item_mod_closures_table_name not in tables:
            op.create_table(
                ticket_item_mod_closures_table_name,
                Column('modifier_key', Integer, primary_key=True),
                Column('target_modifier_key', Integer),
                Column('target_item_key', Integer),
                mysql_charset='utf8',
                mysql_engine='InnoDB',
            )
            with op.batch_alter_table(ticket_item_mod_closures_table_name) as batch_op:
                batch_op.create_foreign_key(
                    'fk_{}_item_modifiers'.format(ticket_item_mod_closures_table_name),
                    ticket_item_mods_table_name,
                    ['modifier_key'],
                    ['key'],
                    onupdate='cascade',
                    ondelete='cascade',
                )
                batch_op.create_foreign_key(
                    'fk_{}_item_modifier_targets'.format(ticket_item_mod_closures_table_name),
                    ticket_item_mods_table_name,
                    ['target_modifier_key'],
                    ['key'],
                    onupdate='cascade',
                    ondelete='cascade',
                )
                batch_op.create_foreign_key(
                    'fk_{}_item_targets'.format(ticket_item_mod_closures_table_name),
                    ticket_items_table_name,
                    ['target_item_key'],
                    ['key'],
                    onupdate='cascade',
                    ondelete='cascade',
                )
                batch_op.create_index(
                    'idx_{}target_modifier_key'.format(maybe_held),
                    ('target_modifier_key',),
                )
                batch_op.create_index(
                    'idx_{}target_item_key'.format(maybe_held),
                    ('target_item_key',),
                )

        if ticket_item_discounts_table_name not in tables:
            op.create_table(
                ticket_item_discounts_table_name,
                Column('ticket_item_key', Integer, primary_key=True),
                Column('id', String(32), primary_key=True),
                Column('comment', Text),
                Column('name', String(128)),
                Column('value', Integer, nullable=False),
                Column('discount_id', String(32)),
                mysql_charset='utf8',
                mysql_engine='InnoDB',
            )
            with op.batch_alter_table(ticket_item_discounts_table_name) as batch_op:
                batch_op.create_foreign_key(
                    'fk_{}_ticket_items'.format(ticket_item_discounts_table_name),
                    ticket_items_table_name,
                    ['ticket_item_key'],
                    ['key'],
                    onupdate='cascade',
                    ondelete='cascade',
                )

        with op.batch_alter_table(tickets_table_name) as batch_op:
            batch_op.drop_column('totals')
            batch_op.drop_column('payments')
            batch_op.drop_column('discounts')
            batch_op.drop_column('service_charges')
            batch_op.drop_column('items')
            batch_op.drop_column('voided_items')

        # Data migration

        ticket_totals_table = table(
            totals_table_name,
            column('ticket_id', String(32)),
            column('discounts', Integer),
            column('due', Integer),
            column('items', Integer),
            column('other_charges', Integer),
            column('paid', Integer),
            column('service_charges', Integer),
            column('tax', Integer),
            column('tips', Integer),
            column('total', Integer),
            column('sub', Integer),
        )
        payments_table = table(
            payments_table_name,
            column('ticket_id', String(32)),
            column('id', String(32)),
            column('type', String(32)),
            column('amount', Integer),
            column('tip', Integer),
            column('change', Integer),
            column('comment', Text),
            column('last4', String(4)),
            column('tender_type_id', String(32)),
        )
        ticket_discounts_table = table(
            discounts_table_name,
            column('ticket_id', String(32)),
            column('id', String(32)),
            column('comment', Text),
            column('name', String(128)),
            column('discount_id', String(32)),
            column('value', Integer),
        )
        ticket_service_charges_table = table(
            service_charges_table_name,
            column('ticket_id', String(32)),
            column('id', String(32)),
            column('name', String(128)),
            column('price', Integer),
            column('comment', Text),
            column('service_charge_id', String(32)),
        )
        item_discounts_table = table(
            ticket_item_discounts_table_name,
            column('ticket_item_key', Integer),
            column('id', String(32)),
            column('comment', Text),
            column('name', String(128)),
            column('value', Integer),
            column('discount_id', String(32)),
        )
        item_modifiers_table = table(
            ticket_item_mods_table_name,
            column('key', Integer),
            column('id', String(32)),
            column('name', String(128)),
            column('quantity', Integer),
            column('split', Integer),
            column('price', Integer),
            column('comment', Text),
            column('sent', Boolean),
            column('sent_at', DateTime),
            column('menu_modifier_id', String(32)),
        )
        items_table = table(
            ticket_items_table_name,
            column('key', Integer),
            column('ticket_id', String(32)),
            column('id', String(32)),
            column('name', String(128)),
            column('quantity', Integer),
            column('split', Integer),
            column('price', Integer),
            column('comment', Text),
            column('sent', Boolean),
            column('sent_at', DateTime),
            column('void', Boolean),
            column('menu_item_id', String(32)),
        )
        closure_table = table(
            ticket_item_mod_closures_table_name,
            column('modifier_key', Integer),
            column('target_modifier_key', Integer),
            column('target_item_key', Integer),
        )
        # SELECT from backup database, INSERT into production database
        row_count = backup_engine.execute("SELECT COUNT(1) FROM {}".format(tickets_table_name)).scalar()
        chunk_size = 5000
        for offset in get_offsets(row_count, chunk_size=chunk_size):
            logger.info("Upgrading {maybe_held}tickets {start} to {finish}...".format(
                maybe_held=maybe_held,
                start=offset,
                finish=min(offset + chunk_size, row_count),
            ))
            rows = backup_engine.execute(
                " SELECT id, totals, payments, discounts, service_charges, items, voided_items "
                " FROM {table_name} "
                " LIMIT {limit} "
                " OFFSET {offset} ".format(
                    table_name=tickets_table_name,
                    limit=chunk_size,
                    offset=offset,
                )
            ).fetchall()
            ticket_totals = [
                (
                    row['id'],
                    pickle.loads(row['totals']) if row['totals'] is not None else TicketTotals()
                )
                for row in rows
            ]
            # Coerce data to not throw given non-nullable columns
            for ticket_id, totals in ticket_totals:
                for attr in (
                    "discounts",
                    "due",
                    "items",
                    "other_charges",
                    "paid",
                    "service_charges",
                    "tax",
                    "tips",
                    "total",
                    "sub",
                ):
                    if getattr(totals, attr) is None:
                        setattr(totals, attr, 0)
            tickets_payments = [
                (
                    row['id'],
                    pickle.loads(row['payments']) if row['payments'] is not None else []
                )
                for row in rows
            ]
            tickets_discounts = [
                (
                    row['id'],
                    pickle.loads(row['discounts']) if row['discounts'] is not None else []
                )
                for row in rows
            ]
            tickets_service_charges = [
                (
                    row['id'],
                    pickle.loads(row['service_charges']) if row['service_charges'] is not None else []
                )
                for row in rows
            ]
            tickets_items = [
                (
                    row['id'],
                    pickle.loads(row['items']) if row['items'] is not None else [],
                    pickle.loads(row['voided_items']) if row['voided_items'] is not None else []
                )
                for row in rows
            ]

            op.bulk_insert(
                ticket_totals_table,
                [
                    {
                        'ticket_id': ticket_id,
                        'discounts': ticket_total.discounts,
                        'due': ticket_total.due,
                        'items': ticket_total.items,
                        'other_charges': ticket_total.other_charges,
                        'paid': ticket_total.paid,
                        'service_charges': ticket_total.service_charges,
                        'tax': ticket_total.tax,
                        'tips': ticket_total.tips,
                        'total': ticket_total.total,
                        'sub': ticket_total.sub,
                    }
                    for ticket_id, ticket_total in ticket_totals
                ]
            )
            op.bulk_insert(
                payments_table,
                [
                    {
                        'ticket_id': ticket_id,
                        'id': ticket_payment.id,
                        'type': ticket_payment.type,
                        'amount': ticket_payment.amount,
                        'tip': ticket_payment.tip,
                        'change': ticket_payment.change,
                        'comment': ticket_payment.comment,
                        'last4': ticket_payment.last4,
                        'tender_type_id': ticket_payment.tender_type_id,
                    }
                    for ticket_id, ticket_payments in tickets_payments
                    for ticket_payment in ticket_payments
                ]
            )
            op.bulk_insert(
                ticket_discounts_table,
                [
                    {
                        'ticket_id': ticket_id,
                        'id': ticket_discount.id,
                        'comment': ticket_discount.comment,
                        'name': ticket_discount.name,
                        'discount_id': ticket_discount.discount_id,
                        'value': ticket_discount.value,
                    }
                    for ticket_id, ticket_discounts in tickets_discounts
                    for ticket_discount in ticket_discounts
                ]
            )
            op.bulk_insert(
                ticket_service_charges_table,
                [
                    {
                        'ticket_id': ticket_id,
                        'id': ticket_service_charge.id,
                        'name': ticket_service_charge.name,
                        'price': ticket_service_charge.price,
                        'comment': ticket_service_charge.comment if hasattr(ticket_service_charge, 'comment') else None,
                        'service_charge_id': ticket_service_charge.service_charge_id if hasattr(ticket_service_charge, 'service_charge_id') else None,
                    }
                    for ticket_id, ticket_service_charges in tickets_service_charges
                    for ticket_service_charge in ticket_service_charges
                ]
            )

            def _persist_item_modifier(connection, modifier, target_item_key=None, target_modifier_key=None):
                insert_mod_result = connection.execute(
                    item_modifiers_table
                    .insert()
                    .values(
                        {
                            'id': modifier.id,
                            'name': modifier.name,
                            'quantity': modifier.quantity,
                            'split': modifier.split,
                            'price': modifier.price,
                            'comment': modifier.comment,
                            'sent': modifier.sent,
                            'sent_at': modifier.sent_at,
                            'menu_modifier_id': modifier.menu_modifier_id,
                        }
                    )
                )
                new_mod_key = (
                    insert_mod_result.inserted_primary_key[0]
                    if insert_mod_result.inserted_primary_key
                    else insert_mod_result.lastrowid
                )

                connection.execute(
                    closure_table
                    .insert()
                    .values(
                        {
                            'modifier_key': new_mod_key,
                            'target_modifier_key': target_modifier_key,
                            'target_item_key': target_item_key,
                        }
                    )
                )

                return new_mod_key

            def _persist_mods_of_mod(connection, modifier, target_modifier_key):
                for mod in modifier.modifiers:
                    new_modifier_key = _persist_item_modifier(connection, mod, target_modifier_key=target_modifier_key)
                    _persist_mods_of_mod(connection, mod, new_modifier_key)

            def _persist_ticket_item(ticket_id, item, void):
                item_insert_result = production_connection.execute(
                    items_table
                    .insert()
                    .values(
                        {
                            'ticket_id': ticket_id,
                            'id': item.id,
                            'name': item.name,
                            'quantity': item.quantity,
                            'split': item.split,
                            'price': item.price,
                            'comment': item.comment,
                            'sent': item.sent,
                            'sent_at': item.sent_at,
                            'void': void,
                            'menu_item_id': item.menu_item_id,
                        }
                    )
                )
                ticket_item_key = (
                    item_insert_result.inserted_primary_key[0]
                    if item_insert_result.inserted_primary_key
                    else item_insert_result.lastrowid
                )

                if item.discounts:
                    production_connection.execute(
                        item_discounts_table.insert(),
                        *(
                            {
                                'ticket_item_key': ticket_item_key,
                                'id': discount.id,
                                'comment': discount.comment,
                                'name': discount.name,
                                'value': discount.value,
                                'discount_id': discount.discount_id,
                            }
                            for discount in item.discounts
                        )
                    )

                for mod in item.modifiers:
                    new_modifier_key = _persist_item_modifier(production_connection, mod, target_item_key=ticket_item_key)
                    _persist_mods_of_mod(production_connection, mod, target_modifier_key=new_modifier_key)

            with production_connection.begin():
                for ticket_id, items, voided_items in tickets_items:
                    for item in items:
                        _persist_ticket_item(ticket_id, item, False)
                    for voided_item in voided_items:
                        _persist_ticket_item(ticket_id, voided_item, True)


def downgrade():
    """
    Use batch-operations to support sqlite

    > with op.batch_alter_table('table_name') as batch_op:
    >     pass
    """
    production_connection = op.get_bind()

    production_filename = production_connection.engine.url.database.replace('sqlite:///', '', 1)
    backup_filename = make_cache_backup_filename(production_filename, revision)
    shutil.copy(production_filename, backup_filename)
    backup_engine = create_engine('sqlite:///' + backup_filename)

    try:
        _downgrade(production_connection, backup_engine)
    except:
        logger.exception(
            "Downgrade from revision %s failed. "
            "You can restore the cache from backup file %s after upgrading.",
            revision,
            backup_filename
        )
        production_connection.close()
        backup_engine.dispose()
        raise
    else:
        logger.info('Downgrade from revision %s succeeded.', revision)
        backup_engine.dispose()


@deinstrument_classes_during_call
def _downgrade(production_connection, backup_engine):
    """
    Use batch-operations to support sqlite

    > with op.batch_alter_table('table_name') as batch_op:
    >     pass
    """
    # Add columns to menu_items
    with op.batch_alter_table('menu_items') as batch_op:
        batch_op.add_column(Column('price_levels', PickleType))
        batch_op.add_column(Column('option_sets', PickleType))

    # Add columns to modifiers
    with op.batch_alter_table('modifiers') as batch_op:
        batch_op.add_column(Column('price_levels', PickleType))
        batch_op.add_column(Column('option_sets', PickleType))

    # Drop old tables
    op.drop_table('option_sets')
    op.drop_table('price_levels')

    def _make_price_level_from_row(row):
        pl = PriceLevel()
        pl.id = str(row[1]) if row[1] is not None else None
        pl.name = str(row[2]) if row[2] is not None else None
        pl.price_per_unit = row[3]
        return remove_sa_instance_state(pl)
    def _make_option_set_from_row(row):
        option_set = OptionSet()
        option_set.id = str(row[1]) if row[1] is not None else None
        option_set.menu_modifier_group_id = str(row[2]) if row[2] is not None else None
        option_set.required = bool(row[3])
        option_set.minimum = row[4]
        option_set.maximum = row[5]
        return remove_sa_instance_state(option_set)

    menu_items_table = table(
        'menu_items',
        column('id', String(32)),
        column('pos_id', String(32)),
        column('name', String(128)),
        column('price_per_unit', Integer),
        column('open', Boolean),
        column('in_stock', Boolean),
        column('price_levels', PickleType),
        column('option_sets', PickleType),
        column('_stub', Boolean),
    )
    modifiers_table = table(
        'modifiers',
        column('id', String(32)),
        column('pos_id', String(32)),
        column('name', String(128)),
        column('price_per_unit', Integer),
        column('open', Boolean),
        column('in_stock', Boolean),
        column('price_levels', PickleType),
        column('option_sets', PickleType),
        column('_stub', Boolean),
    )

    # SELECT from backup database, INSERT into production database
    for menu_objects_table, foreign_key in [
        (menu_items_table, 'menu_item_id'),
        (modifiers_table, 'menu_modifier_id'),
    ]:
        row_count = backup_engine.execute("SELECT COUNT(1) FROM {}".format(menu_objects_table.name)).scalar()
        chunk_size = 5000

        for offset in get_offsets(row_count, chunk_size=chunk_size):
            id_rows = backup_engine.execute(
                " SELECT id "
                " FROM {table_name} "
                " ORDER BY id "
                " LIMIT {limit} "
                " OFFSET {offset} ".format(
                    table_name=menu_objects_table.name,
                    limit=chunk_size,
                    offset=offset,
                )
            ).fetchall()
            menu_object_ids = [id_row['id'] for id_row in id_rows]

            price_level_rows = backup_engine.execute(
                " SELECT {foreign_key}, id, name, price_per_unit "
                " FROM price_levels "
                " WHERE "
                " ( "
                "   ( {foreign_key} IS NOT NULL ) "
                "   AND "
                "   ( {foreign_key} IN ({menu_object_ids}) ) "
                " ) ".format(
                    foreign_key=foreign_key,
                    menu_object_ids=", ".join(map(quote, menu_object_ids))
                )
            ).fetchall()
            keyed_price_levels = [
                (str(row[foreign_key]), _make_price_level_from_row(row))
                for row in price_level_rows
            ]
            grouped_menu_object_price_levels = defaultdict(list)
            for menu_object_id, price_level in keyed_price_levels:
                grouped_menu_object_price_levels[menu_object_id].append(price_level)

            with production_connection.begin() as transaction:
                for menu_object_id in menu_object_ids:
                    production_connection.execute(
                        menu_objects_table
                        .update()
                        .where(menu_objects_table.c.id == menu_object_id)
                        .values({'price_levels': grouped_menu_object_price_levels[menu_object_id]})
                    )

            option_set_rows = backup_engine.execute(
                " SELECT {foreign_key}, id, menu_modifier_group_id, required, minimum, maximum "
                " FROM option_sets "
                " WHERE "
                " ( "
                "   ( {foreign_key} IS NOT NULL ) "
                "   AND "
                "   ( {foreign_key} IN ({menu_object_ids}) ) "
                " ) ".format(
                    foreign_key=foreign_key,
                    menu_object_ids=", ".join(map(quote, menu_object_ids))
                )
            ).fetchall()
            keyed_option_sets = [
                (str(row[foreign_key]), _make_option_set_from_row(row))
                for row in option_set_rows
            ]
            grouped_menu_object_option_sets = defaultdict(list)
            for menu_object_id, option_set in keyed_option_sets:
                grouped_menu_object_option_sets[menu_object_id].append(option_set)

            with production_connection.begin() as transaction:
                for menu_object_id in menu_object_ids:
                    production_connection.execute(
                        menu_objects_table
                        .update()
                        .where(menu_objects_table.c.id == menu_object_id)
                        .values({'option_sets': grouped_menu_object_option_sets[menu_object_id]})
                    )

    for maybe_held in (
        '',  # historical tickets
        'held_'  # held tickets
    ):
        # ALTER production database
        tickets_table_name = maybe_held + 'tickets'
        totals_table_name = maybe_held + 'ticket_totals'
        payments_table_name = maybe_held + 'payments'
        discounts_table_name = maybe_held + 'ticket_discounts'
        service_charges_table_name = maybe_held + 'ticket_service_charges'
        ticket_items_table_name = maybe_held + 'ticket_items'
        ticket_item_mods_table_name = maybe_held + 'ticket_item_modifiers'
        ticket_item_mod_closures_table_name = maybe_held + 'ticket_item_modifier_closures'
        ticket_item_discounts_table_name = maybe_held + 'ticket_item_discounts'

        # Add columns to tickets
        with op.batch_alter_table(tickets_table_name) as batch_op:
            batch_op.add_column(Column('discounts', PickleType))
            batch_op.add_column(Column('items', PickleType))
            batch_op.add_column(Column('voided_items', PickleType))
            batch_op.add_column(Column('payments', PickleType))
            batch_op.add_column(Column('service_charges', PickleType))
            batch_op.add_column(Column('totals', PickleType))

        # Drop old tables
        op.drop_table(totals_table_name)
        op.drop_table(payments_table_name)
        op.drop_table(discounts_table_name)
        op.drop_table(service_charges_table_name)
        op.drop_table(ticket_item_mod_closures_table_name)
        op.drop_table(ticket_item_mods_table_name)
        op.drop_table(ticket_item_discounts_table_name)
        op.drop_table(ticket_items_table_name)

        # Migrate data
        def _make_ticket_totals_from_row(row):
            ttls = TicketTotals()
            ttls.discounts = row[1]
            ttls.due = row[2]
            ttls.items = row[3]
            ttls.other_charges = row[4]
            ttls.paid = row[5]
            ttls.service_charges = row[6]
            ttls.tax = row[7]
            ttls.tips = row[8]
            ttls.total = row[9]
            ttls.sub = row[10]
            return remove_sa_instance_state(ttls)
        def _make_payment_from_row(row):
            pmnt = Payment()
            pmnt.id = str(row[1]) if row[1] is not None else None
            pmnt.type = str(row[2]) if row[2] is not None else None
            pmnt.amount = row[3]
            pmnt.tip = row[4]
            pmnt.change = row[5]
            pmnt.comment = str(row[6]) if row[6] is not None else None
            pmnt.last4 = str(row[7]) if row[7] is not None else None
            pmnt.tender_type_id = str(row[8]) if row[8] is not None else None
            return remove_sa_instance_state(pmnt)
        def _make_ticket_discount_from_row(row):
            dsc = TicketDiscount()
            dsc.id = str(row[1]) if row[1] is not None else None
            dsc.comment = str(row[2]) if row[2] is not None else None
            dsc.name = str(row[3]) if row[3] is not None else None
            dsc.discount_id = str(row[4]) if row[4] is not None else None
            dsc.value = row[5]
            return remove_sa_instance_state(dsc)
        def _make_ticket_service_charge_from_row(row):
            svc_chg = TicketServiceCharge()
            svc_chg.id = str(row[1]) if row[1] is not None else None
            svc_chg.name = str(row[2]) if row[2] is not None else None
            svc_chg.price = row[3]
            svc_chg.comment = str(row[4]) if row[4] is not None else None
            svc_chg.service_charge_id = str(row[5]) if row[5] is not None else None
            return remove_sa_instance_state(svc_chg)
        def _make_item_from_row(row):
            item = TicketItem()
            item.id = str(row['id']) if row['id'] is not None else None
            item.name = str(row['name']) if row['name'] is not None else None
            item.quantity = row['quantity']
            item.split = row['split']
            item.is_split = item.split and item.split > 1
            item.price = row['price']
            item.comment = str(row['comment']) if row['comment'] is not None else None
            item.sent = bool(row['sent']) if row['sent'] is not None else None
            item.sent_at = datetime.datetime.strptime(row['sent_at'], '%Y-%m-%d %H:%M:%S.%f') if row['sent_at'] is not None else None
            item.menu_item_id = str(row['menu_item_id']) if row['menu_item_id'] is not None else None
            return remove_sa_instance_state(item)
        def _make_item_discount_from_row(row):
            dsc = TicketItemDiscount()
            dsc.id = str(row['id']) if row['id'] is not None else None
            dsc.comment = str(row['comment']) if row['comment'] is not None else None
            dsc.name = str(row['name']) if row['name'] is not None else None
            dsc.value = row['value']
            dsc.discount_id = str(row['discount_id']) if row['discount_id'] is not None else None
            return remove_sa_instance_state(dsc)
        def _make_modifier_from_row(row):
            mod = TicketItemModifier()
            mod.id = str(row['id']) if row['id'] is not None else None
            mod.name = str(row['name']) if row['name'] is not None else None
            mod.quantity = row['quantity']
            mod.split = row['split']
            mod.is_split = mod.split and mod.split > 1
            mod.price = row['price']
            mod.comment = str(row['comment']) if row['comment'] is not None else None
            mod.sent = bool(row['sent']) if row['sent'] is not None else None
            mod.sent_at = datetime.datetime.strptime(row['sent_at'], '%Y-%m-%d %H:%M:%S.%f') if row['sent_at'] is not None else None
            mod.menu_modifier_id = str(row['menu_modifier_id']) if row['menu_modifier_id'] is not None else None
            return remove_sa_instance_state(mod)

        tickets_table = table(
            tickets_table_name,
            column('id', String(32)),
            column('ticket_number', Integer),
            column('name', String(128)),
            column('guest_count', Integer),
            column('open', Boolean),
            column('opened_at', DateTime),
            column('closed_at', DateTime),
            column('auto_send', Boolean),
            column('void', Boolean),
            column('table_id', String(32)),
            column('revenue_center_id', String(32)),
            column('employee_id', String(32)),
            column('order_type_id', String(32)),
            column('discounts', PickleType),
            column('items', PickleType),
            column('voided_items', PickleType),
            column('payments', PickleType),
            column('service_charges', PickleType),
            column('totals', PickleType),
        )

        # SELECT from backup database, INSERT into production database
        row_count = backup_engine.execute("SELECT COUNT(1) FROM {}".format(tickets_table_name)).scalar()

        # One query that we don't want to batch
        modifier_closure_rows = backup_engine.execute(
            " SELECT modifier_key, target_modifier_key, target_item_key "
            " FROM {ticket_item_mod_closures_table_name} ".format(
                ticket_item_mod_closures_table_name=ticket_item_mod_closures_table_name
            )
        ).fetchall()
        mod_keys_by_target_item_key = defaultdict(list)
        for row in modifier_closure_rows:
            if row['target_item_key'] is not None:
                mod_keys_by_target_item_key[row['target_item_key']].append(row['modifier_key'])
        mod_keys_by_target_mod_key = defaultdict(list)
        for row in modifier_closure_rows:
            if row['target_modifier_key'] is not None:
                mod_keys_by_target_mod_key[row['target_modifier_key']].append(row['modifier_key'])

        chunk_size = 5000
        for offset in get_offsets(row_count, chunk_size=chunk_size):
            logger.info("Downgrading {maybe_held}tickets {start} to {finish}...".format(
                maybe_held=maybe_held,
                start=offset,
                finish=offset + chunk_size,
            ))
            id_rows = backup_engine.execute(
                " SELECT id "
                " FROM {table_name} "
                " ORDER BY id "
                " LIMIT {limit} "
                " OFFSET {offset} ".format(
                    table_name=tickets_table_name,
                    limit=chunk_size,
                    offset=offset,
                )
            ).fetchall()
            ticket_ids = [id_row['id'] for id_row in id_rows]

            ticket_totals_rows = backup_engine.execute(
                " SELECT ticket_id, discounts, due, items, other_charges, paid, service_charges, tax, tips, total, sub "
                " FROM {totals_table_name} "
                " WHERE ticket_id IN ({ticket_ids}) ".format(
                    totals_table_name=totals_table_name,
                    ticket_ids=", ".join(map(quote, ticket_ids)),
                )
            ).fetchall()
            totals_map = defaultdict(
                lambda: None,
                {
                    str(row['ticket_id']): _make_ticket_totals_from_row(row)
                    for row in ticket_totals_rows
                }
            )
            with production_connection.begin() as transaction:
                for ticket_id in ticket_ids:
                    production_connection.execute(
                        tickets_table
                        .update()
                        .where(tickets_table.c.id == ticket_id)
                        .values({'totals': totals_map[ticket_id]})
                    )

            payments_rows = backup_engine.execute(
                " SELECT ticket_id, id, type, amount, tip, change, comment, last4, tender_type_id "
                " FROM {payments_table_name} "
                " WHERE ticket_id IN ({ticket_ids}) ".format(
                    payments_table_name=payments_table_name,
                    ticket_ids=", ".join(map(quote, ticket_ids)),
                )
            ).fetchall()
            payments = [
                (str(row['ticket_id']), _make_payment_from_row(row))
                for row in payments_rows
            ]
            payments_map = defaultdict(list)
            for ticket_id, payment in payments:
                payments_map[ticket_id].append(payment)
            with production_connection.begin() as transaction:
                for ticket_id in ticket_ids:
                    production_connection.execute(
                        tickets_table
                        .update()
                        .where(tickets_table.c.id == ticket_id)
                        .values({'payments': payments_map[ticket_id]})
                    )

            ticket_discounts_rows = backup_engine.execute(
                " SELECT ticket_id, id, comment, name, discount_id, value "
                " FROM {discounts_table_name} "
                " WHERE ticket_id IN ({ticket_ids}) ".format(
                    discounts_table_name=discounts_table_name,
                    ticket_ids=", ".join(map(quote, ticket_ids)),
                )
            ).fetchall()
            ticket_discounts = [
                (str(row['ticket_id']), _make_ticket_discount_from_row(row))
                for row in ticket_discounts_rows
            ]
            discounts_map = defaultdict(list)
            for ticket_id, discount in ticket_discounts:
                discounts_map[ticket_id].append(discount)
            with production_connection.begin() as transaction:
                for ticket_id in ticket_ids:
                    production_connection.execute(
                        tickets_table
                        .update()
                        .where(tickets_table.c.id == ticket_id)
                        .values({'discounts': discounts_map[ticket_id]})
                    )

            ticket_service_charges_rows = backup_engine.execute(
                " SELECT ticket_id, id, name, price, comment, service_charge_id "
                " FROM {service_charges_table_name} "
                " WHERE ticket_id IN ({ticket_ids}) ".format(
                    service_charges_table_name=service_charges_table_name,
                    ticket_ids=", ".join(map(quote, ticket_ids)),
                )
            ).fetchall()
            ticket_service_charges = [
                (str(row['ticket_id']), _make_ticket_service_charge_from_row(row))
                for row in ticket_service_charges_rows
            ]
            service_charges_map = defaultdict(list)
            for ticket_id, service_charge in ticket_service_charges:
                service_charges_map[ticket_id].append(service_charge)
            with production_connection.begin() as transaction:
                for ticket_id in ticket_ids:
                    production_connection.execute(
                        tickets_table
                        .update()
                        .where(tickets_table.c.id == ticket_id)
                        .values({'service_charges': service_charges_map[ticket_id]})
                    )

            items_rows = backup_engine.execute(
                " SELECT ticket_id, key, void, id, name, quantity, split, price, comment, sent, sent_at, menu_item_id "
                " FROM {ticket_items_table_name} "
                " WHERE ticket_id IN ({ticket_ids}) ".format(
                    ticket_items_table_name=ticket_items_table_name,
                    ticket_ids=", ".join(map(quote, ticket_ids)),
                )
            ).fetchall()
            all_items = {
                row['key']: (str(row['ticket_id']), bool(row['void']), _make_item_from_row(row))
                for row in items_rows
            }

            item_discounts_rows = backup_engine.execute(
                " SELECT ticket_item_key, id, comment, name, value, discount_id "
                " FROM {ticket_item_discounts_table_name} "
                " WHERE ticket_item_key IN ({all_items_keys}) ".format(
                    ticket_item_discounts_table_name=ticket_item_discounts_table_name,
                    all_items_keys=', '.join(quote(str(k)) for k in all_items.keys())
                )
            ).fetchall()
            item_discounts = (
                (row['ticket_item_key'], _make_item_discount_from_row(row))
                for row in item_discounts_rows
            )
            for item_key, discount in item_discounts:
                all_items[item_key][2].discounts.append(discount)

            # mods of items
            mod_keys_to_fetch = [
                mod_key
                for ticket_item_key in all_items.keys()
                for mod_key in mod_keys_by_target_item_key[ticket_item_key]
            ]
            # mods of mods
            mods_of_mods_keys_to_fetch = [
                mod_key
                for target_mod_key in mod_keys_to_fetch
                for mod_key in mod_keys_by_target_mod_key[target_mod_key]
            ]
            while mods_of_mods_keys_to_fetch:
                mod_keys_to_fetch.extend(mods_of_mods_keys_to_fetch)
                mods_of_mods_keys_to_fetch = [
                    mod_key
                    for target_mod_key in mods_of_mods_keys_to_fetch
                    for mod_key in mod_keys_by_target_mod_key[target_mod_key]
                ]
            modifiers_rows = backup_engine.execute(
                " SELECT "
                " closures.target_modifier_key as target_modifier_key, closures.target_item_key as target_item_key,"
                " mods.key as key, mods.id as id, mods.name as name, mods.quantity as quantity, mods.split as split, "
                " mods.price as price, mods.comment as comment, mods.sent as sent, mods.sent_at as sent_at, "
                " mods.menu_modifier_id as menu_modifier_id "
                " "
                " FROM {ticket_item_mods_table_name} as mods "
                " JOIN {ticket_item_mod_closures_table_name} as closures "
                " ON mods.key = closures.modifier_key "
                " WHERE key IN ({mod_keys_to_fetch}) ".format(
                    ticket_item_mods_table_name=ticket_item_mods_table_name,
                    ticket_item_mod_closures_table_name=ticket_item_mod_closures_table_name,
                    mod_keys_to_fetch=', '.join(quote(str(k)) for k in mod_keys_to_fetch),
                )
            ).fetchall()
            grouped_modifiers = defaultdict(list)
            for row in modifiers_rows:
                grouped_modifiers[
                    (row['target_item_key'], row['target_modifier_key'])
                ].append(
                    (row['key'], _make_modifier_from_row(row))
                )

            def _get_mods_of(target_item_key=None, target_mod_key=None):
                next_level_of_keyed_mods = grouped_modifiers[(target_item_key, target_mod_key)]
                for mod_key, mod in next_level_of_keyed_mods:
                    mod.modifiers = _get_mods_of(target_mod_key=mod_key)
                return [mod for _, mod in next_level_of_keyed_mods]

            for item_key, values in all_items.items():
                item = values[2]
                item.modifiers = _get_mods_of(target_item_key=item_key)

            grouped_items = defaultdict(list)
            grouped_voided_items = defaultdict(list)
            for item_key, values in all_items.items():
                ticket_id, is_void, item = values
                if is_void:
                    grouped_voided_items[ticket_id].append(item)
                else:
                    grouped_items[ticket_id].append(item)

            with production_connection.begin() as transaction:
                for ticket_id in ticket_ids:
                    production_connection.execute(
                        tickets_table
                        .update()
                        .where(tickets_table.c.id == ticket_id)
                        .values(
                            {
                                'items': grouped_items[ticket_id],
                                'voided_items': grouped_voided_items[ticket_id],
                            }
                        )
                    )


@deinstrument_classes_during_call
def _gateway_upgrade(engine):
    # CREATE new tables
    print("Creating table option_sets...")
    engine.execute(
        """
        CREATE TABLE IF NOT EXISTS `option_sets` (
          `_cache_id` varchar(128) COLLATE utf8_bin NOT NULL,
          `key` int(11) NOT NULL AUTO_INCREMENT,
          `menu_item_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
          `menu_modifier_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
          `id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
          `menu_modifier_group_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
          `required` tinyint(1) DEFAULT NULL,
          `minimum` int(11) DEFAULT NULL,
          `maximum` int(11) DEFAULT NULL,
          PRIMARY KEY (`key`,`_cache_id`),
          KEY `idx_option_sets_menu_item_id_id` (`_cache_id`,`menu_item_id`,`id`),
          KEY `idx_option_sets_menu_modifier_id_id` (`_cache_id`,`menu_modifier_id`,`id`),
          CONSTRAINT `fk_option_sets_menu_modifier_group` FOREIGN KEY (`_cache_id`, `menu_modifier_group_id`) REFERENCES `modifier_groups` (`_cache_id`, `id`) ON DELETE CASCADE ON UPDATE CASCADE,
          CONSTRAINT `fk_option_sets_menu_item` FOREIGN KEY (`_cache_id`, `menu_item_id`) REFERENCES `menu_items` (`_cache_id`, `id`) ON DELETE CASCADE ON UPDATE CASCADE,
          CONSTRAINT `fk_option_sets_menu_modifier` FOREIGN KEY (`_cache_id`, `menu_modifier_id`) REFERENCES `modifiers` (`_cache_id`, `id`) ON DELETE CASCADE ON UPDATE CASCADE
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
        """
    )
    option_sets_table = table(
        'option_sets',
        Column('_cache_id', String(128)),
        Column('key', Integer),
        Column('menu_item_id', String(32)),
        Column('menu_modifier_id', String(32)),
        Column('id', String(32)),
        Column('menu_modifier_group_id', String(32)),
        Column('required', Boolean),
        Column('minimum', Integer),
        Column('maximum', Integer),
    )

    print("Creating table price_levels...")
    engine.execute(
        """
        CREATE TABLE IF NOT EXISTS `price_levels` (
          `_cache_id` varchar(128) COLLATE utf8_bin NOT NULL,
          `key` int(11) NOT NULL AUTO_INCREMENT,
          `menu_item_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
          `menu_modifier_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
          `id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
          `name` varchar(128) COLLATE utf8_bin DEFAULT NULL,
          `price_per_unit` int(11) DEFAULT NULL,
          PRIMARY KEY (`key`,`_cache_id`),
          KEY `idx_price_levels_menu_item_id_id` (`_cache_id`,`menu_item_id`,`id`),
          KEY `idx_price_levels_menu_modifier_id_id` (`_cache_id`,`menu_modifier_id`,`id`),
          CONSTRAINT `fk_price_levels_menu_item` FOREIGN KEY (`_cache_id`, `menu_item_id`) REFERENCES `menu_items` (`_cache_id`, `id`) ON DELETE CASCADE ON UPDATE CASCADE,
          CONSTRAINT `fk_price_levels_menu_modifier` FOREIGN KEY (`_cache_id`, `menu_modifier_id`) REFERENCES `modifiers` (`_cache_id`, `id`) ON DELETE CASCADE ON UPDATE CASCADE
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
        """
    )
    price_levels_table = table(
        'price_levels',
        Column('_cache_id', String(128)),
        Column('key', Integer),
        Column('menu_item_id', String(32)),
        Column('menu_modifier_id', String(32)),
        Column('id', String(32)),
        Column('name', String(128)),
        Column('price_per_unit', Integer),
    )

    # Data migration
    # SELECT from Pickle columns, INSERT into new tables
    for table_name, foreign_key_column in [
        ('menu_items', 'menu_item_id'),
        ('modifiers', 'menu_modifier_id'),
    ]:
        row_count = engine.execute("SELECT COUNT(1) FROM {}".format(table_name)).scalar()
        chunk_size = 5000
        for offset in get_offsets(row_count, chunk_size=chunk_size):
            print("Upgrading {table_name} option_sets, price_levels for objects {start} to {finish}...".format(
                table_name=table_name,
                start=offset,
                finish=min(offset + chunk_size, row_count),
            ))
            menu_object_rows = engine.execute(
                " SELECT _cache_id, id, price_levels, option_sets "
                " FROM {table_name} "
                " LIMIT {limit} "
                " OFFSET {offset} ".format(
                    table_name=table_name,
                    limit=chunk_size,
                    offset=offset,
                )
            ).fetchall()
            price_levels = [
                (
                    row['_cache_id'],
                    row['id'],
                    pickle.loads(row['price_levels']) if row['price_levels'] is not None else []
                )
                for row in menu_object_rows
            ]
            option_sets = [
                (
                    row['_cache_id'],
                    row['id'],
                    pickle.loads(row['option_sets']) if row['option_sets'] is not None else []
                )
                for row in menu_object_rows
            ]
            price_level_values = [
                {
                    "_cache_id": cache_id,
                    foreign_key_column: foreign_key_value,
                    "id": price_level.id,
                    "name": price_level.name,
                    "price_per_unit": price_level.price_per_unit,
                }
                for cache_id, foreign_key_value, price_lvls in price_levels
                for price_level in price_lvls
            ]
            if any(price_level_values):
                engine.execute(
                    price_levels_table.insert(),
                    *price_level_values
                )
            option_set_values = [
                {
                    "_cache_id": cache_id,
                    foreign_key_column: foreign_key_value,
                    "id": option_set.id,
                    "menu_modifier_group_id": option_set.menu_modifier_group_id,
                    "required": option_set.required,
                    "minimum": option_set.minimum,
                    "maximum": option_set.maximum,
                }
                for cache_id, foreign_key_value, op_sets in option_sets
                for option_set in op_sets
            ]
            if any(option_set_values):
                engine.execute(
                    option_sets_table.insert(),
                    *option_set_values
                )

    # DROP old columns
    _drop_column_if_exists('menu_items', 'price_levels', engine)
    _drop_column_if_exists('menu_items', 'option_sets', engine)
    _drop_column_if_exists('modifiers', 'price_levels', engine)
    _drop_column_if_exists('modifiers', 'option_sets', engine)

    for maybe_held in (
        '',  # historical tickets
        'held_'  # held tickets
    ):
        # CREATE new tables
        tickets_table_name = maybe_held + 'tickets'
        totals_table_name = maybe_held + 'ticket_totals'
        payments_table_name = maybe_held + 'payments'
        discounts_table_name = maybe_held + 'ticket_discounts'
        service_charges_table_name = maybe_held + 'ticket_service_charges'
        ticket_items_table_name = maybe_held + 'ticket_items'
        ticket_item_mods_table_name = maybe_held + 'ticket_item_modifiers'
        ticket_item_mod_closures_table_name = maybe_held + 'ticket_item_modifier_closures'
        ticket_item_discounts_table_name = maybe_held + 'ticket_item_discounts'

        engine.execute(
            """
            CREATE TABLE IF NOT EXISTS`{totals_table_name}` (
              `_cache_id` varchar(128) COLLATE utf8_bin NOT NULL,
              `ticket_id` varchar(32) COLLATE utf8_bin NOT NULL,
              `discounts` int(11) NOT NULL,
              `due` int(11) NOT NULL,
              `items` int(11) NOT NULL,
              `other_charges` int(11) NOT NULL,
              `paid` int(11) NOT NULL,
              `service_charges` int(11) NOT NULL,
              `tax` int(11) NOT NULL,
              `tips` int(11) NOT NULL,
              `total` int(11) NOT NULL,
              `sub` int(11) NOT NULL,
              PRIMARY KEY (`_cache_id`,`ticket_id`),
              CONSTRAINT `fk_{totals_table_name}_tickets` FOREIGN KEY (`_cache_id`, `ticket_id`) REFERENCES `{tickets_table_name}` (`_cache_id`, `id`) ON DELETE CASCADE ON UPDATE CASCADE
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
            """.format(
                totals_table_name=totals_table_name,
                tickets_table_name=tickets_table_name,
            )
        )
        ticket_totals_table = table(
            totals_table_name,
            Column('_cache_id', String(128)),
            column('ticket_id', String(32)),
            column('discounts', Integer),
            column('due', Integer),
            column('items', Integer),
            column('other_charges', Integer),
            column('paid', Integer),
            column('service_charges', Integer),
            column('tax', Integer),
            column('tips', Integer),
            column('total', Integer),
            column('sub', Integer),
        )

        engine.execute(
            """
            CREATE TABLE IF NOT EXISTS`{payments_table_name}` (
              `_cache_id` varchar(128) COLLATE utf8_bin NOT NULL,
              `ticket_id` varchar(32) COLLATE utf8_bin NOT NULL,
              `id` varchar(32) COLLATE utf8_bin NOT NULL,
              `type` varchar(32) COLLATE utf8_bin NOT NULL,
              `amount` int(11) NOT NULL,
              `tip` int(11) NOT NULL,
              `change` int(11) DEFAULT NULL,
              `comment` text COLLATE utf8_bin,
              `last4` varchar(4) COLLATE utf8_bin DEFAULT NULL,
              `tender_type_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
              PRIMARY KEY (`_cache_id`,`ticket_id`,`id`),
              CONSTRAINT `fk_{payments_table_name}_tickets` FOREIGN KEY (`_cache_id`, `ticket_id`) REFERENCES `{tickets_table_name}` (`_cache_id`, `id`) ON DELETE CASCADE ON UPDATE CASCADE
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
            """.format(
                payments_table_name=payments_table_name,
                tickets_table_name=tickets_table_name,
            )
        )
        payments_table = table(
            payments_table_name,
            Column('_cache_id', String(128)),
            column('ticket_id', String(32)),
            column('id', String(32)),
            column('type', String(32)),
            column('amount', Integer),
            column('tip', Integer),
            column('change', Integer),
            column('comment', Text),
            column('last4', String(4)),
            column('tender_type_id', String(32)),
        )

        engine.execute(
            """
            CREATE TABLE IF NOT EXISTS`{discounts_table_name}` (
              `_cache_id` varchar(128) COLLATE utf8_bin NOT NULL,
              `ticket_id` varchar(32) COLLATE utf8_bin NOT NULL,
              `id` varchar(32) COLLATE utf8_bin NOT NULL,
              `comment` text COLLATE utf8_bin,
              `name` varchar(128) COLLATE utf8_bin DEFAULT NULL,
              `discount_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
              `value` int(11) NOT NULL,
              PRIMARY KEY (`_cache_id`,`ticket_id`,`id`),
              CONSTRAINT `fk_{discounts_table_name}_tickets` FOREIGN KEY (`_cache_id`, `ticket_id`) REFERENCES `{tickets_table_name}` (`_cache_id`, `id`) ON DELETE CASCADE ON UPDATE CASCADE
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
            """.format(
                discounts_table_name=discounts_table_name,
                tickets_table_name=tickets_table_name,
            )
        )
        ticket_discounts_table = table(
            discounts_table_name,
            Column('_cache_id', String(128)),
            column('ticket_id', String(32)),
            column('id', String(32)),
            column('comment', Text),
            column('name', String(128)),
            column('discount_id', String(32)),
            column('value', Integer),
        )

        engine.execute(
            """
            CREATE TABLE IF NOT EXISTS`{service_charges_table_name}` (
              `_cache_id` varchar(128) COLLATE utf8_bin NOT NULL,
              `ticket_id` varchar(32) COLLATE utf8_bin NOT NULL,
              `id` varchar(32) COLLATE utf8_bin NOT NULL,
              `name` varchar(128) COLLATE utf8_bin DEFAULT NULL,
              `price` int(11) NOT NULL,
              `comment` text COLLATE utf8_bin,
              `service_charge_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
              PRIMARY KEY (`_cache_id`,`ticket_id`,`id`),
              CONSTRAINT `fk_{service_charges_table_name}_tickets` FOREIGN KEY (`_cache_id`, `ticket_id`) REFERENCES `{tickets_table_name}` (`_cache_id`, `id`) ON DELETE CASCADE ON UPDATE CASCADE
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
            """.format(
                service_charges_table_name=service_charges_table_name,
                tickets_table_name=tickets_table_name,
            )
        )
        ticket_service_charges_table = table(
            service_charges_table_name,
            Column('_cache_id', String(128)),
            column('ticket_id', String(32)),
            column('id', String(32)),
            column('name', String(128)),
            column('price', Integer),
            column('comment', Text),
            column('service_charge_id', String(32)),
        )

        engine.execute(
            """
            CREATE TABLE IF NOT EXISTS`{ticket_item_mods_table_name}` (
              `_cache_id` varchar(128) COLLATE utf8_bin NOT NULL,
              `key` int(11) NOT NULL AUTO_INCREMENT,
              `id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
              `name` varchar(128) COLLATE utf8_bin DEFAULT NULL,
              `quantity` int(11) NOT NULL,
              `split` int(11) DEFAULT NULL,
              `price` int(11) NOT NULL,
              `comment` text COLLATE utf8_bin,
              `sent` tinyint(1) DEFAULT NULL,
              `sent_at` datetime DEFAULT NULL,
              `menu_modifier_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
              PRIMARY KEY (`key`,`_cache_id`)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
            """.format(
                ticket_item_mods_table_name=ticket_item_mods_table_name,
            )
        )
        item_modifiers_table = table(
            ticket_item_mods_table_name,
            Column('_cache_id', String(128)),
            column('key', Integer),
            column('id', String(32)),
            column('name', String(128)),
            column('quantity', Integer),
            column('split', Integer),
            column('price', Integer),
            column('comment', Text),
            column('sent', Boolean),
            column('sent_at', DateTime),
            column('menu_modifier_id', String(32)),
        )

        engine.execute(
            """
            CREATE TABLE IF NOT EXISTS`{ticket_items_table_name}` (
              `_cache_id` varchar(128) COLLATE utf8_bin NOT NULL,
              `key` int(11) NOT NULL AUTO_INCREMENT,
              `ticket_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
              `id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
              `name` varchar(128) COLLATE utf8_bin DEFAULT NULL,
              `quantity` int(11) NOT NULL,
              `split` int(11) DEFAULT NULL,
              `price` int(11) NOT NULL,
              `comment` text COLLATE utf8_bin,
              `sent` tinyint(1) DEFAULT NULL,
              `sent_at` datetime DEFAULT NULL,
              `void` tinyint(1) NOT NULL,
              `menu_item_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
              PRIMARY KEY (`key`,`_cache_id`),
              UNIQUE KEY `idx_{ticket_or_held_ticket}_id_id` (`_cache_id`,`ticket_id`,`id`),
              CONSTRAINT `fk_{ticket_items_table_name}_tickets` FOREIGN KEY (`_cache_id`, `ticket_id`) REFERENCES `{tickets_table_name}` (`_cache_id`, `id`) ON DELETE CASCADE ON UPDATE CASCADE
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
            """.format(
                ticket_items_table_name=ticket_items_table_name,
                tickets_table_name=tickets_table_name,
                ticket_or_held_ticket=tickets_table_name[:-1]
            )
        )
        items_table = table(
            ticket_items_table_name,
            Column('_cache_id', String(128)),
            column('key', Integer),
            column('ticket_id', String(32)),
            column('id', String(32)),
            column('name', String(128)),
            column('quantity', Integer),
            column('split', Integer),
            column('price', Integer),
            column('comment', Text),
            column('sent', Boolean),
            column('sent_at', DateTime),
            column('void', Boolean),
            column('menu_item_id', String(32)),
        )

        engine.execute(
            """
            CREATE TABLE IF NOT EXISTS`{ticket_item_mod_closures_table_name}` (
              `_cache_id` varchar(128) COLLATE utf8_bin DEFAULT NULL,
              `modifier_key` int(11) NOT NULL,
              `target_modifier_key` int(11) DEFAULT NULL,
              `target_item_key` int(11) DEFAULT NULL,
              PRIMARY KEY (`modifier_key`),
              CONSTRAINT `fk_{ticket_item_mod_closures_table_name}_item_modifiers` FOREIGN KEY (`modifier_key`, `_cache_id`) REFERENCES `{ticket_item_mods_table_name}` (`key`, `_cache_id`) ON DELETE CASCADE ON UPDATE CASCADE,
              CONSTRAINT `fk_{ticket_item_mod_closures_table_name}_item_modifier_targets` FOREIGN KEY (`target_modifier_key`, `_cache_id`) REFERENCES `{ticket_item_mods_table_name}` (`key`, `_cache_id`) ON DELETE CASCADE ON UPDATE CASCADE,
              CONSTRAINT `fk_{ticket_item_mod_closures_table_name}_item_targets` FOREIGN KEY (`target_item_key`, `_cache_id`) REFERENCES `{ticket_items_table_name}` (`key`, `_cache_id`) ON DELETE CASCADE ON UPDATE CASCADE
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
            """.format(
                ticket_item_mod_closures_table_name=ticket_item_mod_closures_table_name,
                tickets_table_name=tickets_table_name,
                ticket_item_mods_table_name=ticket_item_mods_table_name,
                ticket_items_table_name=ticket_items_table_name,
            )
        )
        closure_table = table(
            ticket_item_mod_closures_table_name,
            Column('_cache_id', String(128)),
            column('modifier_key', Integer),
            column('target_modifier_key', Integer),
            column('target_item_key', Integer),
        )

        engine.execute(
            """
            CREATE TABLE IF NOT EXISTS`{ticket_item_discounts_table_name}` (
              `_cache_id` varchar(128) COLLATE utf8_bin NOT NULL,
              `ticket_item_key` int(11) NOT NULL,
              `id` varchar(32) COLLATE utf8_bin NOT NULL,
              `comment` text COLLATE utf8_bin,
              `name` varchar(128) COLLATE utf8_bin DEFAULT NULL,
              `value` int(11) NOT NULL,
              `discount_id` varchar(32) COLLATE utf8_bin DEFAULT NULL,
              PRIMARY KEY (`_cache_id`,`ticket_item_key`,`id`),
              CONSTRAINT `fk_{ticket_item_discounts_table_name}_ticket_items` FOREIGN KEY (`ticket_item_key`, `_cache_id`) REFERENCES `{ticket_items_table_name}` (`key`, `_cache_id`) ON DELETE CASCADE ON UPDATE CASCADE
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
            """.format(
                ticket_item_discounts_table_name=ticket_item_discounts_table_name,
                ticket_items_table_name=ticket_items_table_name,
            )
        )
        item_discounts_table = table(
            ticket_item_discounts_table_name,
            Column('_cache_id', String(128)),
            column('ticket_item_key', Integer),
            column('id', String(32)),
            column('comment', Text),
            column('name', String(128)),
            column('value', Integer),
            column('discount_id', String(32)),
        )

        # Data migration
        # SELECT from Pickle columns, INSERT into new tables
        row_count = engine.execute("SELECT COUNT(1) FROM {}".format(tickets_table_name)).scalar()
        chunk_size = 5000
        for offset in get_offsets(row_count, chunk_size=chunk_size):
            print("Upgrading {maybe_held}tickets {start} to {finish}...".format(
                maybe_held=maybe_held,
                start=offset,
                finish=min(offset + chunk_size, row_count),
            ))
            rows = engine.execute(
                " SELECT `_cache_id`, id, totals, payments, discounts, service_charges, items, voided_items "
                " FROM {table_name} "
                " LIMIT {limit} "
                " OFFSET {offset} ".format(
                    table_name=tickets_table_name,
                    limit=chunk_size,
                    offset=offset,
                )
            ).fetchall()
            ticket_totals = [
                (
                    row['_cache_id'],
                    row['id'],
                    pickle.loads(row['totals']) if row['totals'] is not None else TicketTotals()
                )
                for row in rows
            ]
            # Coerce data to not throw given non-nullable columns
            for cache_id, ticket_id, totals in ticket_totals:
                for attr in (
                    "discounts",
                    "due",
                    "items",
                    "other_charges",
                    "paid",
                    "service_charges",
                    "tax",
                    "tips",
                    "total",
                    "sub",
                ):
                    if getattr(totals, attr) is None:
                        setattr(totals, attr, 0)
            tickets_payments = [
                (
                    row['_cache_id'],
                    row['id'],
                    pickle.loads(row['payments']) if row['payments'] is not None else []
                )
                for row in rows
            ]
            tickets_discounts = [
                (
                    row['_cache_id'],
                    row['id'],
                    pickle.loads(row['discounts']) if row['discounts'] is not None else []
                )
                for row in rows
            ]
            tickets_service_charges = [
                (
                    row['_cache_id'],
                    row['id'],
                    pickle.loads(row['service_charges']) if row['service_charges'] is not None else []
                )
                for row in rows
            ]
            tickets_items = [
                (
                    row['_cache_id'],
                    row['id'],
                    pickle.loads(row['items']) if row['items'] is not None else [],
                    pickle.loads(row['voided_items']) if row['voided_items'] is not None else []
                )
                for row in rows
            ]
            ticket_totals_values = [
                {
                    "_cache_id": cache_id,
                    "ticket_id": ticket_id,
                    "discounts": ticket_total.discounts,
                    "due": ticket_total.due,
                    "items": ticket_total.items,
                    "other_charges": ticket_total.other_charges,
                    "paid": ticket_total.paid,
                    "service_charges": ticket_total.service_charges,
                    "tax": ticket_total.tax,
                    "tips": ticket_total.tips,
                    "total": ticket_total.total,
                    "sub": ticket_total.sub,
                }
                for cache_id, ticket_id, ticket_total in ticket_totals
            ]
            if any(ticket_totals_values):
                engine.execute(
                    ticket_totals_table.insert(),
                    *ticket_totals_values
                )
            payments_values = [
                {
                    "_cache_id": cache_id,
                    "ticket_id": ticket_id,
                    "id": ticket_payment.id,
                    "type": ticket_payment.type,
                    "amount": ticket_payment.amount,
                    "tip": ticket_payment.tip,
                    "change": ticket_payment.change,
                    "comment": ticket_payment.comment,
                    "last4": ticket_payment.last4,
                    "tender_type_id": ticket_payment.tender_type_id,
                }
                for cache_id, ticket_id, ticket_payments in tickets_payments
                for ticket_payment in ticket_payments
            ]
            if any(payments_values):
                engine.execute(
                    payments_table.insert(),
                    *payments_values
                )
            ticket_discount_values = [
                {
                    "_cache_id": cache_id,
                    "ticket_id": ticket_id,
                    "id": ticket_discount.id,
                    "comment": ticket_discount.comment,
                    "name": ticket_discount.name,
                    "discount_id": ticket_discount.discount_id,
                    "value": ticket_discount.value,
                }
                for cache_id, ticket_id, ticket_discounts in tickets_discounts
                for ticket_discount in ticket_discounts
            ]
            if any(ticket_discount_values):
                engine.execute(
                    ticket_discounts_table.insert(),
                    *ticket_discount_values
                )
            ticket_service_charge_values = (
                {
                    "_cache_id": cache_id,
                    "ticket_id": ticket_id,
                    "id": ticket_service_charge.id,
                    "name": ticket_service_charge.name,
                    "price": ticket_service_charge.price,
                    "comment": ticket_service_charge.comment,
                    "service_charge_id": ticket_service_charge.service_charge_id,
                }
                for cache_id, ticket_id, ticket_service_charges in tickets_service_charges
                for ticket_service_charge in ticket_service_charges
            )
            if any(ticket_service_charge_values):
                engine.execute(
                    ticket_service_charges_table.insert(),
                    *ticket_service_charge_values
                )

            def _persist_item_modifier(connection, cache_id, modifier, target_item_key=None, target_modifier_key=None):
                insert_mod_result = connection.execute(
                    item_modifiers_table
                    .insert()
                    .values(
                        {
                            '_cache_id': cache_id,
                            'id': modifier.id,
                            'name': modifier.name,
                            'quantity': modifier.quantity,
                            'split': modifier.split,
                            'price': modifier.price,
                            'comment': modifier.comment,
                            'sent': modifier.sent,
                            'sent_at': modifier.sent_at,
                            'menu_modifier_id': modifier.menu_modifier_id,
                        }
                    )
                )
                new_mod_key = (
                    insert_mod_result.inserted_primary_key[1]
                    if insert_mod_result.inserted_primary_key
                    else insert_mod_result.lastrowid
                )

                connection.execute(
                    closure_table
                    .insert()
                    .values(
                        {
                            '_cache_id': cache_id,
                            'modifier_key': new_mod_key,
                            'target_modifier_key': target_modifier_key,
                            'target_item_key': target_item_key,
                        }
                    )
                )

                return new_mod_key

            def _persist_mods_of_mod(connection, cache_id, modifier, target_modifier_key):
                for mod in modifier.modifiers:
                    new_modifier_key = _persist_item_modifier(connection, cache_id, mod, target_modifier_key=target_modifier_key)
                    _persist_mods_of_mod(connection, cache_id, mod, new_modifier_key)

            def _persist_ticket_item(connection, cache_id, ticket_id, item, void):
                item_insert_result = connection.execute(
                    items_table
                    .insert()
                    .values(
                        {
                            '_cache_id': cache_id,
                            'ticket_id': ticket_id,
                            'id': item.id,
                            'name': item.name,
                            'quantity': item.quantity,
                            'split': item.split,
                            'price': item.price,
                            'comment': item.comment,
                            'sent': item.sent,
                            'sent_at': item.sent_at,
                            'void': void,
                            'menu_item_id': item.menu_item_id,
                        }
                    )
                )
                ticket_item_key = (
                    item_insert_result.inserted_primary_key[1]
                    if item_insert_result.inserted_primary_key
                    else item_insert_result.lastrowid
                )

                if item.discounts:
                    connection.execute(
                        item_discounts_table.insert(),
                        *(
                            {
                                '_cache_id': cache_id,
                                'ticket_item_key': ticket_item_key,
                                'id': discount.id,
                                'comment': discount.comment,
                                'name': discount.name,
                                'value': discount.value,
                                'discount_id': discount.discount_id,
                            }
                            for discount in item.discounts
                        )
                    )

                for mod in item.modifiers:
                    new_modifier_key = _persist_item_modifier(connection, cache_id, mod, target_item_key=ticket_item_key)
                    _persist_mods_of_mod(connection, cache_id, mod, target_modifier_key=new_modifier_key)

            with engine.connect() as connection:
                with connection.begin():
                    for cache_id, ticket_id, items, voided_items in tickets_items:
                        for item in items:
                            _persist_ticket_item(connection, cache_id, ticket_id, item, False)
                        for voided_item in voided_items:
                            _persist_ticket_item(connection, cache_id, ticket_id, voided_item, True)

        # DROP old columns
        _drop_column_if_exists(tickets_table_name, 'totals', engine)
        _drop_column_if_exists(tickets_table_name, 'payments', engine)
        _drop_column_if_exists(tickets_table_name, 'discounts', engine)
        _drop_column_if_exists(tickets_table_name, 'service_charges', engine)
        _drop_column_if_exists(tickets_table_name, 'items', engine)
        _drop_column_if_exists(tickets_table_name, 'voided_items', engine)

    print("Migration Complete")


def _drop_column_if_exists(table_name, column_name, engine):
    if _has_column(table_name, column_name, engine):
        print("Dropping {column_name} column from {table_name} table.".format(
            column_name=column_name,
            table_name=table_name,
        ))
        engine.execute(
            """
            ALTER TABLE `{table_name}`
                DROP COLUMN `{column_name}`;
            """.format(
                table_name=table_name,
                column_name=column_name,
            )
        )


def _has_column(table_name, column_name, engine):
    inspector = inspect(engine)
    return column_name in {
        col['name']
        for col in inspector.get_columns(table_name)
    }


if __name__ == "__main__":
    import argparse

    UPGRADE, DOWNGRADE = 'upgrade', 'downgrade'

    parser = argparse.ArgumentParser(description='Apply these migrations to a MySQL database.')
    parser.add_argument(
        'direction',
        action='store',
        choices=(UPGRADE, DOWNGRADE),
        help='The direction in which to run the migrations.',
    )
    parser.add_argument(
        'conn_string',
        action='store',
        help='The connection string to create a SqlAlchemy engine.',
    )
    args = parser.parse_args()

    if args.direction == UPGRADE:
        _gateway_upgrade(create_engine(args.conn_string, echo=False, convert_unicode=True))
    elif args.direction == DOWNGRADE:
        print("Downgrade not implemented for gateway!")
