SchemaMigration

Complexity

Total Complexity

8

Size/Duplication

Total Lines	53
Duplicated Lines	0 %

Test Coverage

Coverage

94.12%

Importance

Changes

0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	run()	0	19	3
A	_build_router()	0	6	2
B	reset()	0	25	3

from plugin.core.database.manager import DatabaseManager
from plugin.models.core import db, migrations_path
from plugin.modules.migrations.core.base import Migration

import logging

log = logging.getLogger(__name__)

# Try import "peewee_migrate" router
try:
    from peewee_migrate.core import Router
except (ImportError, NameError):
    Router = None


class SchemaMigration(Migration):
    def run(self):
        log.debug('migrations_path: %r', migrations_path)

        # Build migration router
        router = self._build_router()

        if not router:
            return False

        # Validate current schema
        if not router.validate():
            log.error('Detected corrupt/invalid database schema, resetting database...')
            return self.reset()
        else:
            log.info('Database schema is valid')

        # Run schema migrations
        router.run()
        return True

    @classmethod
    def reset(cls):
        if not DatabaseManager.reset('main', db, 'invalid-schema'):
            # Unable to reset database
            return False

        # Build migration router
        router = cls._build_router()

        if not router:
            return False

        # Run migrations
        router.run()

        # Log message to channel menu
        from plugin.managers.message import MessageManager
        from plugin.models import Message

        MessageManager.get.from_message(logging.WARNING,

It seems like a value for argument message is missing in the unbound method call.

            "Plugin database has been automatically reset due to schema corruption, see http://bit.ly/TFPx90101 for more details",

This line is too long as per the coding-style (130/120).

            code=Message.Code.DatabaseSchemaCorruptionReset
        )

        return True

    @staticmethod
    def _build_router():
        if not Router:
            return None

        return Router(migrations_path, DATABASE=db)