Source code for PyFunceble.cli.migrators.db_base

"""
The tool to check the availability or syntax of domain, IP or URL.

::


    ██████╗ ██╗   ██╗███████╗██╗   ██╗███╗   ██╗ ██████╗███████╗██████╗ ██╗     ███████╗
    ██╔══██╗╚██╗ ██╔╝██╔════╝██║   ██║████╗  ██║██╔════╝██╔════╝██╔══██╗██║     ██╔════╝
    ██████╔╝ ╚████╔╝ █████╗  ██║   ██║██╔██╗ ██║██║     █████╗  ██████╔╝██║     █████╗
    ██╔═══╝   ╚██╔╝  ██╔══╝  ██║   ██║██║╚██╗██║██║     ██╔══╝  ██╔══██╗██║     ██╔══╝
    ██║        ██║   ██║     ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
    ╚═╝        ╚═╝   ╚═╝      ╚═════╝ ╚═╝  ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝

Provides the base of all our database migration.

Author:
    Nissar Chababy, @funilrys, contactTATAfunilrysTODTODcom

Special thanks:
    https://pyfunceble.github.io/#/special-thanks

Contributors:
    https://pyfunceble.github.io/#/contributors

Project link:
    https://github.com/funilrys/PyFunceble

Project documentation:
    https://pyfunceble.readthedocs.io/en/dev/

Project homepage:
    https://pyfunceble.github.io/

License:
::


    Copyright 2017, 2018, 2019, 2020, 2022 Nissar Chababy

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
"""

import functools
from typing import Any

import PyFunceble.cli.facility
import PyFunceble.cli.factory
import PyFunceble.sessions
from PyFunceble.cli.migrators.base import MigratorBase


[docs]class DBMigratorBase(MigratorBase): """ Provides the base of all our database migration. """ def __init__(self, print_action_to_stdout: bool = False) -> None: super().__init__(print_action_to_stdout=print_action_to_stdout)
[docs] def execute_if_authorized(default: Any = None): # pylint: disable=no-self-argument """ Executes the decorated method only if we are authorized to process. Otherwise, apply the given :code:`default`. """ def inner_metdhod(func): @functools.wraps(func) def wrapper(self, *args, **kwargs): if self.authorized: return func(self, *args, **kwargs) # pylint: disable=not-callable return self if default is None else default return wrapper return inner_metdhod
@property def authorized(self): """ Provides the authorization to run. """ return PyFunceble.cli.facility.CredentialLoader.is_already_loaded()
[docs] def does_table_exists(self, table_name: str) -> bool: """ Checks if the table exists. :param table_name: The name of the table to check. """ statement = ( "SELECT COUNT(*) " "FROM information_schema.tables " "WHERE table_schema = :database_name " "AND table_name = :table_name " ) result = self.db_session.execute( statement, { # pylint: disable=line-too-long "database_name": PyFunceble.cli.facility.CredentialLoader.credential.name, "table_name": table_name, }, ) result = dict(result.fetchone()) if result["COUNT(*)"] != 1: return False return True
[docs] def start(self) -> "MigratorBase": """ Starts the migration. """ raise NotImplementedError()