Source code for anyblok.scripts

# This file is a part of the AnyBlok project
#
#    Copyright (C) 2014 Jean-Sebastien SUZANNE <[email protected]>
#    Copyright (C) 2018 Jean-Sebastien SUZANNE <[email protected]>
#
# This Source Code Form is subject to the terms of the Mozilla Public License,
# v. 2.0. If a copy of the MPL was not distributed with this file,You can
# obtain one at http://mozilla.org/MPL/2.0/.
import anyblok
from anyblok.release import version
from anyblok.blok import BlokManager
from anyblok.config import Configuration, get_db_name
from anyblok.registry import RegistryManager, return_list
from anyblok._graphviz import ModelSchema, SQLSchema
from nose import main
import warnings
import sys
from os.path import join
from os import walk
from argparse import RawDescriptionHelpFormatter
from textwrap import dedent
from sqlalchemy_utils.functions import create_database
from anyblok import (
    load_init_function_from_entry_points,
    configuration_post_load,
)

Configuration.add_application_properties(
    'createdb',
    [
        'unittest',
        'logging',
        'create_db',
        'install-bloks',
        'install-or-update-bloks',
    ],
    prog='AnyBlok create database, version %r' % version,
    description="Create a database and install bloks to populate it"
)

Configuration.add_application_properties(
    'updatedb',
    [
        'unittest',
        'logging',
        'install-bloks',
        'uninstall-bloks',
        'update-bloks',
        'install-or-update-bloks',
    ],
    prog='AnyBlok update database, version %r' % version,
    description="Update a database: install, upgrade or uninstall the bloks "
)

Configuration.add_application_properties(
    'nose', ['logging', 'unittest'],
    prog='AnyBlok nose, version %r' % version,
    description="Run fonctionnal nosetest of the installed bloks"
)

Configuration.add_application_properties(
    'interpreter', ['logging', 'interpreter'],
    prog='AnyBlok interpretor, version %r' % version,
    description="Run an interpreter on the registry",
    formatter_class=RawDescriptionHelpFormatter,
    epilog=dedent("Example\n"
                  "-------\n"
                  "  $ anyblok_interpreter [anyblok arguments] \n"
                  "  $ => registry \n"
                  "  ... <registry> \n\n"
                  "  The interpretor add in the local the registry of "
                  "the selected database \n\n"
                  "Note\n"
                  "----\n"
                  "  if the ipython is in the python path, then "
                  "the interpretor will be an ipyton interpretor")
)

Configuration.add_application_properties(
    'autodoc', ['logging', 'doc', 'schema'],
    prog='AnyBlok auto documentation, version %r' % version,
)


[docs]def anyblok_createdb(): """Create a database and install blok from config""" load_init_function_from_entry_points() Configuration.load('createdb') configuration_post_load() BlokManager.load() db_name = get_db_name() db_template_name = Configuration.get('db_template_name', None) url = Configuration.get('get_url')(db_name=db_name) create_database(url, template=db_template_name) registry = RegistryManager.get(db_name) if registry is None: return if Configuration.get('install_all_bloks'): bloks = registry.System.Blok.list_by_state('uninstalled') else: install_bloks = Configuration.get('install_bloks') or [] iou_bloks = Configuration.get('install_or_update_bloks') or [] bloks = list(set(install_bloks + iou_bloks)) registry.upgrade(install=bloks) registry.commit() registry.close()
[docs]def anyblok_updatedb(): """Update an existing database""" registry = anyblok.start('updatedb', loadwithoutmigration=True) installed_bloks = registry.System.Blok.list_by_state('installed') toupdate_bloks = registry.System.Blok.list_by_state('toupdate') required_install_bloks = [] required_update_bloks = [] for blok in (Configuration.get('install_or_update_bloks') or []): if blok in installed_bloks: required_update_bloks.append(blok) elif blok not in toupdate_bloks: required_install_bloks.append(blok) if Configuration.get('install_all_bloks'): install_bloks = registry.System.Blok.list_by_state('uninstalled') else: install_bloks = Configuration.get('install_bloks') or [] install_bloks = list(set(install_bloks + required_install_bloks)) if Configuration.get('update_all_bloks'): update_bloks = registry.System.Blok.list_by_state('installed') else: update_bloks = Configuration.get('update_bloks') or [] update_bloks = list(set(update_bloks + required_update_bloks)) uninstall_bloks = Configuration.get('uninstall_bloks') if registry: registry.update_blok_list() # case, new blok added registry.upgrade(install=install_bloks, update=update_bloks, uninstall=uninstall_bloks) registry.commit() registry.close()
[docs]def anyblok_nose(): """Run nose unit test for the registry """ warnings.simplefilter('default') registry = anyblok.start('nose', useseparator=True, unittest=True) defaultTest = [] if registry: installed_bloks = registry.System.Blok.list_by_state("installed") selected_bloks = return_list( Configuration.get('selected_bloks')) or installed_bloks unwanted_bloks = return_list( Configuration.get('unwanted_bloks')) or [] defaultTest = [] for blok in installed_bloks: if blok not in selected_bloks or blok in unwanted_bloks: continue startpath = BlokManager.getPath(blok) for root, dirs, _ in walk(startpath): if 'tests' in dirs: defaultTest.append(join(root, 'tests')) registry.close() # free the registry to force create it again sys.exit(main(defaultTest=defaultTest))
[docs]def anyblok_interpreter(): """Execute a script or open an interpreter """ registry = anyblok.start('interpreter') if registry: registry.commit() python_script = Configuration.get('python_script') if python_script: with open(python_script, "r") as fh: exec(fh.read(), None, locals()) else: try: from IPython import embed embed() except ImportError: import code code.interact(local=locals())
[docs]def anyblok2doc(): """Return auto documentation for the registry """ registry = anyblok.start('autodoc') if registry: registry.commit() doc = registry.Documentation() doc.auto_doc() if Configuration.get('doc_format') == 'RST': with open(Configuration.get('doc_output'), 'w') as fp: doc.toRST(fp) elif Configuration.get('doc_format') == 'UML': format_ = Configuration.get('schema_format') name_ = Configuration.get('schema_output') dot = ModelSchema(name_, format=format_) doc.toUML(dot) dot.save() elif Configuration.get('doc_format') == 'SQL': format_ = Configuration.get('schema_format') name_ = Configuration.get('schema_output') dot = SQLSchema(name_, format=format_) doc.toSQL(dot) dot.save()