Source code for anyblok.blok

# -*- coding: utf-8 -*-
# This file is a part of the AnyBlok project
#    Copyright (C) 2014 Jean-Sebastien SUZANNE <>
# 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
from pkg_resources import iter_entry_points
from anyblok.imp import ImportManager
from .logging import log
from anyblok.environment import EnvironmentManager
from time import sleep
from sys import modules
from os.path import dirname
from logging import getLogger
from os.path import join
from datetime import datetime

logger = getLogger(__name__)

[docs]class BlokManagerException(LookupError): """ Simple exception to BlokManager """ def __init__(self, *args, **kwargs): EnvironmentManager.set('current_blok', None) super(BlokManagerException, self).__init__(*args, **kwargs)
[docs]class BlokManager: """ Manage the bloks for one process A blok has a `setuptools` entrypoint, this entry point is defined by the ``entry_points`` attribute in the first load The ``bloks`` attribute is a dict with all the loaded entry points Use this class to import all the bloks in the entrypoint:: BlokManager.load() """ bloks = {} entry_points = None ordered_bloks = [] auto_install = []
[docs] @classmethod def list(cls): """ Return the ordered bloks :rtype: list of blok name ordered by loading """ return cls.ordered_bloks
[docs] @classmethod def has(cls, blok): """ Return True if the blok is loaded :param blok: blok name :rtype: bool """ return blok and blok in cls.ordered_bloks or False
[docs] @classmethod def get(cls, blok): """ Return the loaded blok :param blok: blok name :rtype: blok instance :exception: BlokManagerException """ if not cls.has(blok): raise BlokManagerException('%r not found' % blok) return cls.bloks[blok]
[docs] @classmethod def set(cls, blokname, blok): """ Add a new blok :param blokname: blok name :param blok: blok instance :exception: BlokManagerException """ if cls.has(blokname): raise BlokManagerException('%r already present' % blokname) cls.bloks[blokname] = blok cls.ordered_bloks.append(blokname)
[docs] @classmethod @log(logger, level='debug') def reload(cls): """ Reload the entry points Empty the ``bloks`` dict and use the ``entry_points`` attribute to load bloks :exception: BlokManagerException """ if cls.entry_points is None: raise BlokManagerException( """You must use the ``load`` classmethod before using """ """``reload``""") entry_points = [] entry_points += cls.entry_points cls.unload() cls.load(entry_points=entry_points)
[docs] @classmethod @log(logger, level='debug') def unload(cls): """ Unload all the bloks but not the registry """ cls.bloks = {} cls.ordered_bloks = [] cls.entry_points = None cls.auto_install = [] from .registry import RegistryManager RegistryManager.unload()
@classmethod def get_need_blok_linked_bloks(cls, blok): for required in cls.bloks[blok].required: if not cls.get_need_blok(required): raise BlokManagerException( "Not %s required bloks found" % required) cls.bloks[required].required_by.append(blok) for optional in cls.bloks[blok].optional: if cls.get_need_blok(optional): cls.bloks[optional].optional_by.append(blok) for conditional in cls.bloks[blok].conditional: cls.bloks[conditional].conditional_by.append(blok) for conflicting in cls.bloks[blok].conflicting: cls.bloks[conflicting].conflicting_by.append(blok) @classmethod def get_need_blok(cls, blok): if cls.has(blok): return True if blok not in cls.bloks: return False cls.get_need_blok_linked_bloks(blok) cls.ordered_bloks.append(blok) EnvironmentManager.set('current_blok', blok) if not ImportManager.has(blok): # Import only if not exist don't reload here mod = ImportManager.add(blok) mod.imports() else: mod = ImportManager.get(blok) mod.reload() if cls.bloks[blok].autoinstall: cls.auto_install.append(blok) return True
[docs] @classmethod @log(logger, level='debug') def load(cls, entry_points=('bloks',)): """ Load all the bloks and import them :param entry_points: Use by ``iter_entry_points`` to get the blok :exception: BlokManagerException """ if not entry_points: raise BlokManagerException("The entry_points mustn't be empty") cls.entry_points = entry_points if EnvironmentManager.get('current_blok'): while EnvironmentManager.get('current_blok'): sleep(0.1) EnvironmentManager.set('current_blok', 'start') bloks = [] for entry_point in entry_points: count = 0 for i in iter_entry_points(entry_point): count += 1 blok = i.load() blok.required_by = [] blok.optional_by = [] blok.conditional_by = [] blok.conflicting_by = [] cls.set(, blok) = bloks.append((blok.priority, if not count: raise BlokManagerException( "Invalid bloks group %r" % entry_point) # Empty the ordered blok to reload it depending on the priority cls.ordered_bloks = [] bloks.sort() try: while bloks: blok = bloks.pop(0)[1] cls.get_need_blok(blok) finally: EnvironmentManager.set('current_blok', None)
[docs] @classmethod def getPath(cls, blok): """ Return the path of the blok :param blok: blok name in ``ordered_bloks`` :rtype: absolute path """ blok = cls.get(blok) return dirname(modules[blok.__module__].__file__)
[docs]class Blok: """ Super class for all the bloks define the default value for: * priority: order to load the blok * required: list of the bloks needed to install this blok * optional: list of the bloks to be installed if present in the blok list * conditional: if all the bloks of this list are installed then install this blok """ autoinstall = False priority = 100 required = [] optional = [] conditional = [] conflicting = [] name = None # filled by the BlokManager author = '' logo = '' def __init__(self, registry): self.registry = registry
[docs] @classmethod def import_declaration_module(cls): """ Do the python import for the Declaration of the model or other """
[docs] def update(self, latest_version): """ Call at the installation or update :param latest_version: latest version installed, if the blok have not been installing the latest_version will be None """
[docs] def pre_migration(self, latest_version): """Call at update, before the automigration .. warning:: You can not use the ORM :param latest_version: latest version installed, if the blok have not been installing the latest_version will be None """
[docs] def post_migration(self, latest_version): """Call at update, after the automigration :param latest_version: latest version installed, if the blok have not been installing the latest_version will be None """
[docs] def uninstall(self): """ Call at the uninstallation """
[docs] def load(self): """ Call at the launch of the application """
[docs] def import_file(self, importer_name, model, *file_path, **kwargs): """ Import data file :param importer_name: Name of the importer (need installation of the Blok which have the importer) :param model: Model of the data to import :param \*file_path: relative path of the path in this Blok :param \*\*kwargs: Option for the importer :rtype: return dict of result """ blok_path = BlokManager.getPath( _file = join(blok_path, *file_path)"import %r file: %r", importer_name, _file) Importer = self.registry.get(BlokManager.get_importer(importer_name)) file_to_import = None with open(_file, 'rb') as fp: file_to_import = importer = Importer.insert( model=model, file_to_import=file_to_import, **kwargs) started_at = res = stoped_at = dt = stoped_at - started_at"Create %d entries, Update %d entries (%d.%d sec)", len(res['created_entries']), len(res['updated_entries']), dt.seconds, dt.microseconds) if 'error_found' in res and res['error_found']: for error in res['error_found']: logger.error(error) else: importer.delete() return res