AnyBlok framework

anyblok module

anyblok.start(processName, version='0.3.0', prompt='%(processName)s - %(version)s', argsparse_groups=None, parts_to_load=None, logger=None, useseparator=False)

Function which initialize the application

registry = start('My application',
                 argsparse_groups=['config', 'database'],
                 parts_to_load=['AnyBlok'])
Parameters:
  • processName – Name of the application
  • version – Version of the application
  • prompt – Prompt message for the help
  • argsparse_groups – list of the group of option for argparse
  • parts_to_load – group of blok to load
  • logger – option to configure logging
  • useseparator – boolean, indicate if argsparse option are split betwwen two application
Return type:

registry if the database name is in the configuration

anyblok.declarations module

class anyblok.declarations.Declarations

Represents all the declarations done by the bloks

Warning

This is a global information, during the execution you must use the registry. The registry is the real assembler of the python classes based on the installed bloks

from anyblok import Declarations
class Column(*args, **kwargs)

Column class

This class can’t be instanciated

get_registry_name(model)

Return the registry name of the remote model

Return type:str of the registry name
get_sqlalchemy_mapping(registry, namespace, fieldname, properties)

Return the instance of the real field

Parameters:
  • registry – current registry
  • namespace – name of the model
  • fieldname – name of the field
  • properties – known properties of the model
Return type:

sqlalchemy column instance

get_tablename(registry, model)

Return the table name of the remote model

Return type:str of the table name
must_be_declared_as_attr()

Return True if the column have a foreign key to a remote column

native_type()

Return the native SqlAlchemy type

class Declarations.Core

The Core class is the base of all the AnyBlok models

Add new core model:

@Declarations.register(Declarations.Core)
class Base:
    pass

Remove the core model:

Declarations.unregister(Declarations.Core, 'Base', Base,
                             blok='MyBlok')
classmethod register(parent, name, cls_, **kwargs)

Add new sub registry in the registry

Parameters:
  • parent – Existing declaration
  • name – Name of the new declaration to add it
  • cls – Class Interface to add in the declaration
classmethod unregister(entry, cls_)

Remove the Interface from the registry

Parameters:
  • entry – entry declaration of the model where the cls_ must be removed
  • cls – Class Interface to remove in the declaration
class Declarations.Exception

Adapter to Exception Class

The Exception class is used to define the type of Declarations Exception

Add new Exception type:

@Declarations.register(Declarations.Exception)
class MyException:
    pass

Removing the exception is forbidden because it can be used

exception ArgsParseManagerException

Simple Exception for ArgsParseManager

exception Declarations.Exception.BlokManagerException(*args, **kwargs)

Simple exception to BlokManager

exception Declarations.Exception.DeclarationsException

Simple Exception for Declarations

exception Declarations.Exception.EnvironmentException

Exception for the Environment

exception Declarations.Exception.FieldException

Simple Exception for Field

exception Declarations.Exception.ImportManagerException

Exception for Import Manager

exception Declarations.Exception.MigrationException

Simple Exception class for Migration

exception Declarations.Exception.ModelException

Exception for Model declaration

exception Declarations.Exception.RegistryException

Simple Exception for Registry

exception Declarations.Exception.RegistryManagerException

Simple Exception for Registry

exception Declarations.Exception.ViewException

Exception for View declaration

classmethod Declarations.Exception.register(parent, name, cls_, **kwargs)

add new sub registry in the registry

Parameters:
  • parent – Existing declaration
  • name – Name of the new declaration to add it
  • cls – Class to add in the declaration
Exception:

DeclarationsException

classmethod Declarations.Exception.unregister(entry, cls_)

Forbidden method

Exception:DeclarationsException
class Declarations.Field(*args, **kwargs)

Field class

This class can’t be instanciated

forbid_instance(cls)

Raise an exception if the cls is an instance of this __class__

Parameters:cls – instance of the class
Exception:FieldException
format_label(fieldname)

Return the label for this field

Parameters:fieldname – if no label filled, the fieldname will be capitalized and returned
Return type:the label for this field
get_sqlalchemy_mapping(registry, namespace, fieldname, properties)

Return the instance of the real field

Parameters:
  • registry – current registry
  • namespace – name of the model
  • fieldname – name of the field
  • properties – properties known of the model
Return type:

instance of Field

must_be_declared_as_attr()

Return False, it is the default value

must_be_duplicate_before_added()

Return False, it is the default value

native_type()

Return the native SqlAlchemy type

Exception:FieldException
classmethod register(parent, name, cls_, **kwargs)

add new sub registry in the registry

Parameters:
  • parent – Parent to attach the declaration to
  • name – Name of the new field
  • cls – Class to add in the declaration
Exception:

FieldException

classmethod unregister(child, cls_)

Forbidden method

Exception:FieldException
update_properties(registry, namespace, fieldname, properties)

Update the propertie use to add new column

Parameters:
  • registry – current registry
  • namespace – name of the model
  • fieldname – name of the field
  • properties – properties known to the model
class Declarations.Mixin

The Mixin class are used to define a behaviours on models:

  • Add new mixin class:

    @Declarations.register(Declarations.Mixin)
    class MyMixinclass:
        pass
    
  • Remove a mixin class:

    Declarations.unregister(Declarations.Mixin.MyMixinclass, MyMixinclass)
    
class Declarations.Model

The Model class is used to define or inherit an SQL table.

Add new model class:

@Declarations.register(Declarations.Model)
class MyModelclass:
    pass

Remove a model class:

Declarations.unregister(Declarations.Model.MyModelclass,
                        MyModelclass)

There are three Model families:

  • No SQL Model: These models have got any field, so any table
  • SQL Model:
  • SQL View Model: it is a model mapped with a SQL View, the insert, update delete method are forbidden by the database

Each model has a:

  • registry name: compose by the parent + . + class model name
  • table name: compose by the parent + ‘_’ + class model name

The table name can be overloaded by the attribute tablename. the wanted value are a string (name of the table) of a model in the declaration.

..warning:

Two models can have the same table name, both models are mapped on
the table. But they must have the same column.
classmethod apply_event_listner(registry, namespace, base, properties)

Find the event listener methods in the base to save the namespace and the method in the registry

Parameters:
  • registry – the current registry
  • namespace – the namespace of the model
  • base – One of the base of the model
  • properties – the properties of the model
classmethod apply_hybrid_method(registry, namespace, bases, properties)

Create overload to define the write declaration of sqlalchemy hybrid method, add the overload in the declared bases of the namespace

Parameters:
  • registry – the current registry
  • namespace – the namespace of the model
  • base – One of the base of the model
  • properties – the properties of the model
classmethod apply_view(namespace, tablename, base, registry, properties)

Transform the sqlmodel to view model

Parameters:
  • namespace – Namespace of the model
  • tablename – Name od the table of the model
  • base – Model cls
  • registry – current registry
  • properties – properties of the model
Exception:

MigrationException

Exception:

ViewException

classmethod assemble_callback(registry)

Assemble callback is called to assemble all the Model from the installed bloks

Parameters:registry – registry to update
classmethod declare_field(registry, name, field, namespace, properties)

Declare the field/column/relationship to put in the properties of the model

Parameters:
  • registry – the current registry
  • name – name of the field / column or relationship
  • field – the declaration field / column or relationship
  • namespace – the namespace of the model
  • properties – the properties of the model
classmethod detect_hybrid_method(registry, namespace, base, properties)

Find the sqlalchemy hybrid methods in the base to save the namespace and the method in the registry

Parameters:
  • registry – the current registry
  • namespace – the namespace of the model
  • base – One of the base of the model
  • properties – the properties of the model
classmethod initialize_callback(registry)

initialize callback is called after assembling all entries

This callback updates the database information about

  • Model
  • Column
  • RelationShip
Parameters:registry – registry to update
classmethod insert_in_bases(registry, namespace, bases, properties)

Add in the declared namespaces new base.

Parameters:
  • registry – the current registry
  • namespace – the namespace of the model
  • base – One of the base of the model
  • properties – the properties of the model
classmethod load_namespace_first_step(registry, namespace)

Return the properties of the declared bases for a namespace. This is the first step because some actions need to known all the properties

Parameters:
  • registry – the current registry
  • namespace – the namespace of the model
Return type:

dict of the known properties

classmethod load_namespace_second_step(registry, namespace, realregistryname=None, transformation_properties=None)

Return the bases and the properties of the namespace

Parameters:
  • registry – the current registry
  • namespace – the namespace of the model
  • realregistryname – the name of the model if the namespace is a mixin
Return type:

the list od the bases and the properties

Exception:

ModelException

classmethod register(parent, name, cls_, **kwargs)

add new sub registry in the registry

Parameters:
  • parent – Existing global registry
  • name – Name of the new registry to add it
  • cls – Class Interface to add in registry
classmethod transform_base(registry, namespace, base, properties)

Detect specific declaration which must define by registry

Parameters:
  • registry – the current registry
  • namespace – the namespace of the model
  • base – One of the base of the model
  • properties – the properties of the model
Return type:

new base

classmethod unregister(entry, cls_)

Remove the Interface from the registry

Parameters:
  • entry – entry declaration of the model where the cls_ must be removed
  • cls – Class Interface to remove in registry
class Declarations.RelationShip(*args, **kwargs)

RelationShip class

The RelationShip class is used to define the type of SQL field Declarations

Add a new relation ship type:

@Declarations.register(Declarations.RelationShip)
class Many2one:
    pass

the relationship column are forbidden because the model can be used on the model

apply_instrumentedlist(registry)

Add the InstrumentedList class to replace List class as result of the query

Parameters:registry – current registry
check_existing_remote_model(registry)

Check if the remote model exists

The information of the existance come from the first step of assembling

Exception:FieldException if the model doesn’t exist
define_backref_properties(registry, namespace, properties)

Add in the backref_properties, new property for the backref

Parameters:
  • registry – current registry
  • namespace – name of the model
  • properties – properties known of the model
find_primary_key(properties)

Return the primary key come from the first step property

Parameters:properties – first step properties for the model
Return type:column name of the primary key
Exception:FieldException
format_backref(registry, namespace, properties)

Create the real backref, with the backref string and the backref properties

Parameters:
  • registry – current registry
  • namespace – name of the model
  • properties – properties known of the model
get_registry_name()

Return the registry name of the remote model

Return type:str of the registry name
get_sqlalchemy_mapping(registry, namespace, fieldname, properties)

Return the instance of the real field

Parameters:
  • registry – current registry
  • namespace – name of the model
  • fieldname – name of the field
  • properties – properties known of the model
Return type:

sqlalchemy relation ship instance

get_tablename(registry, model=None)

Return the table name of the remote model

Return type:str of the table name
must_be_declared_as_attr()

Return True, because it is a relationship

classmethod Declarations.add_declaration_type(cls_=None, isAnEntry=False, assemble=None, initialize=None)

Add a declaration type

Parameters:
  • cls – The class object to add as a world of the MetaData
  • isAnEntry – if true the type will be assembled by the registry
  • assemble – name of the method callback to call (classmethod)
  • initialize – name of the method callback to call (classmethod)
Exception:

DeclarationsException

classmethod Declarations.register(parent, cls_=None, **kwargs)

Method to add the blok in the registry under a type of declaration

Parameters:
  • parent – An existing blok class in the Declaration
  • cls_ – The class object to add in the Declaration
Return type:

cls_

Exception:

DeclarationsException

classmethod Declarations.unregister(entry, cls_)

Method to remove the blok from a type of declaration

Parameters:
  • entry – declaration entry of the model where the cls_ must be removed
  • cls_ – The class object to remove from the Declaration
Return type:

cls_

anyblok._argsparse module

class anyblok._argsparse.ArgsParseManager

ArgsParse is used to define the options of the real argparse and its default values. Each application or blok can declare needed options here.

This class stores three attributes:

  • groups: lists of options indexed by part, a part is a ConfigParser group, or a process name
  • labels: if a group has got a label then all the options in group are gathered in a parser group
  • configuration: result of the ArgsParser after loading
classmethod add(group, part='AnyBlok', label=None, function_=None)

Add a function in a part and a group.

The function must have two arguments:

  • parser: the parser instance of argparse
  • default: A dict with the default value

This function is called to know what the options of this must do. You can declare this group:

  • either by calling the add method as a function:

    def foo(parser, default):
        pass
    
    ArgsParseManager.add('create-db', function_=foo)
    
  • or by calling the add method as a decorator:

    @ArgsParseManager.add('create-db')
    def bar(parser, default):
        pass
    

By default the group is unnamed, if you want a named group, you must set the label attribute:

@ArgsParseManager.add('create-db', label="Name of the group")
def bar(parser, default):
    pass
Parameters:
  • part – ConfigParser group or process name
  • group – group is a set of parser option
  • label – If the group has a label then all the functions in the group are put in group parser
  • function – function to add
classmethod get(opt, default=None)

Get a value from the configuration dict after loading

After the loading of the application, all the options are saved in the ArgsParseManager. And all the applications have free access to these options:

from anyblok._argsparse import ArgsParseManager

database = ArgsParseManager.get('dbname')

..warning:

Some options are used as a default value not real value, such
as the dbname
Parameters:
  • opt – name of the option
  • default – default value if the option doesn’t exist
classmethod get_url(dbname=None)

Return an sqlalchemy URL for database

Get the options of the database, the only option which can be overloaded is the name of the database:

url = ArgsParseManager.get_url(dbname='Mydb')
Parameters:dbname – Name of the database
Return type:SqlAlchemy URL
Exception:ArgsParseManagerException
classmethod load(description='AnyBlok :', argsparse_groups=None, parts_to_load=None, useseparator=False)

Load the argparse definition and parse them

Parameters:
  • description – description of argsparse
  • argsparse_groups – list argsparse groupe to load
  • parts_to_load – group of blok to load
  • useseparator – boolean(default False)
classmethod remove(group, function_, part='AnyBlok')

Remove an existing function

If your application inherits some unwanted options from a specific function, you can unlink this function:

def foo(opt, default):
    pass

ArgsParseManager.add('create-db', function_=foo)
ArgsParseManager.remove('create-db', function_=foo)
Parameters:
  • part – ConfigParser group or process name
  • group – group is a set of parser option
  • function – function to add
classmethod remove_label(group, part='AnyBlok')

Remove an existing label

The goal of this function is to remove an existing label of a specific group:

@ArgsParseManager.add('create-db', label="Name of the group")
def bar(parser, defaul):
    pass

ArgsParseManager.remove_label('create-db')
Parameters:
  • part – ConfigParser group or process name
  • group – group is a set of parser option

anyblok._imp module

class anyblok._imp.ImportManager

Use to import the blok or reload the blok imports

Add a blok and imports its modules:

blok = ImportManager.add('my blok')
blok.imports()

Reload the modules of a blok:

if ImportManager.has('my blok'):
    blok = ImportManager.get('my blok')
    blok.reload()
    # import the unimported module
classmethod add(blok)

Store the blok so that we know which bloks to reload if needed

Parameters:blok – name of the blok to add
Return type:loader instance
Exception:ImportManagerException
classmethod get(blok)

Return the module imported for this blok

Parameters:blok – name of the blok to add
Return type:loader instance
Exception:ImportManagerException
classmethod has(blok)

Return True if the blok was imported

Parameters:blok – name of the blok to add
Return type:boolean

anyblok._logging module

class anyblok._logging.Formatter(fmt=None, datefmt=None, style='%')

Bases: logging.Formatter

Define the format for console logging

format(record)

Add color to the message

Parameters:record – logging record instance
Return type:logging record formatted
anyblok._logging.init_logger(level='info', mode='console', filename=None, socket=None, facility=8)

Init the logger output

There are 5 levels of logging * debug * info (default) * warning * error * critical

Example:

from anyblok.log import init_logger
init_logger(level='debug')

A logger can log to:

  • console (default):

    init_logger(mode='console')
    
  • file:

    init_logger(mode='file', filename='my.file.log')
    
  • socket:

    init_logger(mode='socket', socket=('localhost', 1000))
    
  • syslog:

    Example:

    # By socket
    init_logger(mode='syslog', socket('localhost', 514))
    # By UNIX socket
    init_logger(mode='syslog', socket='/dev/log')
    

    the syslog mode define logger facility:

    • LOG_AUTH
    • LOG_AUTHPRIV
    • LOG_CRON
    • LOG_DAEMON
    • LOG_FTP
    • LOG_KERN
    • LOG_LPR
    • LOG_MAIL
    • LOG_NEWS
    • LOG_SYSLOG
    • LOG_USER (default)
    • LOG_UUCP
    • LOG_LOCAL0
    • LOG_LOCAL1
    • LOG_LOCAL2
    • LOG_LOCAL3
    • LOG_LOCAL4
    • LOG_LOCAL5
    • LOG_LOCAL6
    • LOG_LOCAL7

    example:

    init_logger(mode='syslog', socket='/dev/log',
                facility=syslog.LOG_SYSLOG)
    
Parameters:
  • level – level defined by anyblok
  • mode – Output mode
  • filename – Output file
  • socket – Socket or UnixSocket
  • facility
Exception:

Exception

anyblok._logging.log(level='info', withargs=False)

decorator to log the entry of a method

There are 5 levels of logging * debug * info (default) * warning * error * critical

example:

@log()
def foo(...):
    ...
Parameters:
  • level – AnyBlok log level
  • withargs – If True, add args and kwargs in the log message

anyblok.environment module

class anyblok.environment.EnvironmentManager

Manage the Environment for an application

classmethod define_environment_cls(Environment)

Define the class used for the environment

Parameters:Environment – class of environment
Exception:EnvironmentException
environment

alias of ThreadEnvironment

classmethod get(key, default=None)

Load the value of the key in the environment

Parameters:
  • key – the key of the value to load
  • default – return this value if not value loaded for the key
Return type:

the value of the key

Exception:

EnvironmentException

classmethod scoped_function_for_session()

Save the value of the key in the environment

classmethod set(key, value)

Save the value of the key in the environment

Parameters:
  • key – the key of the value to save
  • value – the value to save
Exception:

EnvironmentException

class anyblok.environment.ThreadEnvironment

Use the thread, to get the environment

classmethod getter(key, default)

Get the value of the key in the environment

Parameters:
  • key – the key of the value to retrieve
  • default – return this value if no value loaded for the key
Return type:

the value of the key

scoped_function_for_session = None

No scoped function here because for none value sqlalchemy already uses a thread to save the session

classmethod setter(key, value)

Save the value of the key in the environment

Parameters:
  • key – the key of the value to save
  • value – the value to save

anyblok.blok module

class anyblok.blok.BlokManager

Manage the bloks for one process

A blok has a setuptools entrypoint, this entry point is defined by the bloks_groups 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('AnyBlok')
classmethod get(blok)

Return the loaded blok

Parameters:blok – blok name
Return type:blok instance
Exception:BlokManagerException
classmethod getPath(blok)

Return the path of the blok

Parameters:blok – blok name in ordered_bloks
Return type:absolute path
classmethod has(blok)

Return True if the blok is loaded

Parameters:blok – blok name
Return type:bool
classmethod list()

Return the ordered bloks

Return type:list of blok name ordered by loading
classmethod load(*bloks_groups)

Load all the bloks and import them

Parameters:bloks_groups – Use by iter_entry_points to get the blok
Exception:BlokManagerException
classmethod reload()

Reload the entry points

Empty the bloks dict and use the bloks_groups attribute to load bloks :exception: BlokManagerException

classmethod set(blokname, blok)

Add a new blok

Parameters:
  • blokname – blok name
  • blok – blok instance
Exception:

BlokManagerException

classmethod unload()

Unload all the bloks but not the registry

class anyblok.blok.Blok(registry)

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
  • conditionnal: if all the bloks of this list are installed then install this blok

anyblok.registry module

class anyblok.registry.RegistryManager

Manage the global registry

Add new entry:

RegistryManager.declare_entry('newEntry')
RegistryManager.init_blok('newBlok')
EnvironmentManager.set('current_blok', 'newBlok')
RegistryManager.add_entry_in_register(
    'newEntry', 'oneKey', cls_)
EnvironmentManager.set('current_blok', None)

Remove an existing entry:

if RegistryManager.has_entry_in_register('newBlok', 'newEntry',
                                                'oneKey'):
    RegistryManager.remove_entry_in_register(
        'newBlok', 'newEntry', 'oneKey', cls_)

get a new registry for a database:

registry = RegistryManager.get('my database')
classmethod add_core_in_register(core, cls_)

Load core in blok

warning the global var current_blok must be filled on the good blok

Parameters:
  • core – is the existing core name
  • cls_ – Class of the Core to save in loaded blok target registry
classmethod add_entry_in_register(entry, key, cls_, **kwargs)

Load entry in blok

warning the global var current_blok must be filled on the good blok :param entry: is the existing entry name :param key: is the existing key in the entry :param cls_: Class of the entry / key to remove in loaded blok

classmethod add_or_replace_blok_property(property_, value)

Save the value in the properties

Parameters:
  • property_ – name of the property
  • value – the value to save, the type is not important
classmethod clear()

Clear the registry dict to force the creation of new registry

classmethod declare_core(core)

Add new core in the declared cores

RegistryManager.declare_core('Core name')

-----------------------------------------

@Declarations.register(Declarations.Core)
class ``Core name``:
    ...

Warning

The core must be declared in the application, not in the bloks The declaration must be done before the loading of the bloks

Parameters:core – core name
classmethod declare_entry(entry, assemble_callback=None, initialize_callback=None)

Add new entry in the declared entries

def assemble_callback(registry):
    ...

def initialize_callback(registry):
    ...

RegistryManager.declare_entry(
    'Entry name', assemble_callback=assemble_callback,
    initialize_callback=initialize_callback)

@Declarations.register(Declarations.``Entry name``)
class MyClass:
    ...

Warning

The entry must be declared in the application, not in the bloks The declaration must be done before the loading of the bloks

Parameters:
  • entry – entry name
  • assemble_callback – function callback to call to assemble
  • initialize_callback – function callback to call to init after assembling
classmethod get(dbname)

Return an existing Registry

If the Registry doesn’t exist then the Registry are created and added to registries dict

Parameters:dbname – the name of the database linked to this registry
Return type:Registry
classmethod get_blok_property(property_, default=None)

Return the value in the properties

Parameters:
  • property_ – name of the property
  • default – return default If not entry in the property
classmethod has_blok(blok)

Return True if the blok is already loaded

Parameters:blok – name of the blok
Return type:boolean
classmethod has_blok_property(property_)

Return True if the property exists in blok

Parameters:property_ – name of the property
classmethod has_core_in_register(blok, core)

Return True if One Class exist in this blok for this core

Parameters:
  • blok – name of the blok
  • core – is the existing core name
classmethod has_entry_in_register(blok, entry, key)

Return True if One Class exist in this blok for this entry

Parameters:
  • blok – name of the blok
  • entry – is the existing entry name
  • key – is the existing key in the entry
classmethod init_blok(blokname)

init one blok to be known by the RegistryManager

All bloks loaded must be initialized because the registry will be created with this information

Parameters:blokname – name of the blok
classmethod reload(blok)

Reload the blok

The purpose is to reload the python module to get changes in python file

Parameters:blok – the name of the blok to reload
classmethod remove_blok_property(property_)

Remove the property if exist

Parameters:property_ – name of the property
classmethod remove_in_register(cls_)

Remove Class in blok and in entry

Parameters:cls_ – Class of the entry / key to remove in loaded blok
class anyblok.registry.Registry(dbname)

Define one registry

A registry is linked to a database, and stores the definition of the installed Bloks, Models, Mixins for this database:

registry = Registry('My database')
add_in_registry(namespace, base)

Add a class as an attribute of the registry

Parameters:
  • namespace – tree path of the attribute
  • base – class to add
clean_model()

Clean the registry of all the namespaces

close()

Release the session, connection and engine

close_session()

Close only the session, not the registry After the call of this method the registry won’t be usable you should use close method which call this method

commit(*args, **kwargs)

Overload the commit method of the SqlAlchemy session

get(namespace)

Return the namespace Class

Parameters:namespace – namespace to get from the registry str
Return type:namespace cls
Exception:RegistryManagerException
get_bloks_by_states(*states)

Return the bloks in these states

Parameters:states – list of the states
Return type:list of blok’s name
get_bloks_to_install(loaded)

Return the bloks to install in the registry

Return type:list of blok’s name
get_bloks_to_load()

Return the bloks to load by the registry

Return type:list of blok’s name
ini_var()

Initialize the var to load the registry

load()

Load all the namespaces of the registry

Create all the table, make the shema migration Update Blok, Model, Column rows

load_blok(blok, toinstall, toload)

load on blok, load all the core and all the entry for one blok

Parameters:blok – name of the blok
Exception:RegistryManagerException
load_core(blok, core)

load one core type for one blok

Parameters:
  • blok – name of the blok
  • core – the core name to load
load_entry(blok, entry)

load one entry type for one blok

Parameters:
  • blok – name of the blok
  • entry – declaration type to load
precommit_hook(registryname, method, put_at_the_if_exist)

Add a method in the precommit_hook list

a precommit hook is a method called just after the commit, it is used to call this method once, because a hook is saved only once

Parameters:
  • registryname – namespace of the model
  • method – method to call on the registryname
  • put_at_the_if_exist – if true and hook allready exist then the hook are moved at the end
reload()

Reload the registry, close session, clean registry, reinit var

upgrade(install=None, update=None, uninstall=None)

Upgrade the current registry

Parameters:
  • install – list of the blok to install
  • update – list of the blok to update
  • uninstall – list of the blok to uninstall
Exception:

RegistryException

anyblok.migration module

Warning

AnyBlok use Alembic to do the dynamic migration, but Alembic does’nt detect all the change (Foreifn key, primary key), we must wait the Alembic or implement it in Alembic project before use it in AnyBlok

class anyblok.migration.MigrationReport(migration, diffs)

Change report

Get a new report:

report = MigrationReport(migrationinstance, change_detected)
apply_change()

Apply the migration

this method parses the detected change and calls the Migration system to apply the change with the api of Declarations

log_has(log)

return True id the log is present

Warning

this method is only used for the unittest

Parameters:log – log sentence expected
class anyblok.migration.MigrationConstraintForeignKey(column)

Used to apply a migration on a foreign key

You can add:

table.column('my column').foreign_key().add(Blok.name)

Or drop:

table.column('my column').foreign_key().drop()
add(remote_field, **kwargs)

Add a new foreign key

Parameters:remote_field – The column of the remote model
Return type:MigrationConstraintForeignKey instance
drop()

Drop the foreign key

class anyblok.migration.MigrationColumn(table, name)

get or add a column

Add a new column:

table.column().add(Sqlachemy column)

Get a column:

c = table.column('My column name')

Alter the column:

c.alter(new_column_name='Another column name')

Drop the column:

c.drop()
add(column)

Add a new column

The column is added in two phases, the last phase is only for the the nullable, if nullable can not be applied, a warning is logged

Parameters:column – sqlalchemy column
Return type:MigrationColumn instance
alter(**kwargs)

Alter an existing column

Alter the column in two phases, because the nullable column has not locked the migration

Warning

See Alembic alter_column, the existing_* param are used for some dialect like mysql, is importante to filled them for these dialect

Parameters:
  • new_column_name – New name for the column
  • type – New sqlalchemy type
  • existing_type – Old sqlalchemy type
  • server_default – The default value in database server
  • existing_server_default – Old default value
  • nullable – New nullable value
  • existing_nullable – Old nullable value
  • autoincrement – New auto increment use for Integer whith primary key only
  • existing_autoincrement – Old auto increment
Return type:

MigrationColumn instance

drop()

Drop the column

foreign_key()

Get a foreign key

Return type:MigrationConstraintForeignKey instance
nullable()

Use for unittest return if the column is nullable

server_default()

Use for unittest: return the default database value

type()

Use for unittest: return the column type

class anyblok.migration.MigrationConstraintCheck(table, name)

Used for the Check constraint

Add a new constraint:

table('My table name').check().add('check_my_column', 'mycolumn > 5')

Get and drop the constraint:

table('My table name').check('check_my_column').drop()
add(name, condition)

Add the constraint

Parameters:
  • name – name of the constraint
  • condition – constraint to apply
Return type:

MigrationConstraintCheck instance

drop()

Drop the constraint

class anyblok.migration.MigrationConstraintUnique(table, *columns, **kwargs)

Used for the Unique constraint

Add a new constraint:

table('My table name').unique().add('col1', 'col2')

Get and drop the constraint:

table('My table name').unique('col1', 'col2').drop()
add(*columns)

Add the constraint

Parameters:*column – list of column name
Return type:MigrationConstraintUnique instance
Exception:MigrationException
drop()

Drop the constraint

class anyblok.migration.MigrationConstraintPrimaryKey(table)

Used for the primary key constraint

Add a new constraint:

table('My table name').primarykey().add('col1', 'col2')

Get and drop the constraint:

table('My table name').primarykey('col1', 'col2').drop()
add(*columns)

Add the constraint

Parameters:*column – list of column name
Return type:MigrationConstraintPrimaryKey instance
Exception:MigrationException
drop()

Drop the constraint

class anyblok.migration.MigrationIndex(table, *columns, **kwargs)

Used for the index constraint

Add a new constraint:

table('My table name').index().add('col1', 'col2')

Get and drop the constraint:

table('My table name').index('col1', 'col2').drop()
add(*columns)

Add the constraint

Parameters:*column – list of column name
Return type:MigrationIndex instance
Exception:MigrationException
drop()

Drop the constraint

class anyblok.migration.MigrationTable(migration, name)

Use to manipulate tables

Add a table:

table().add('New table')

Get an existing table:

t = table('My table name')

Alter the table:

t.alter(name='Another table name')

Drop the table:

t.drop()
add(name)

Add a new table

Parameters:name – name of the table
Return type:MigrationTable instance
alter(**kwargs)

Atler the current table

Parameters:name – New table name
Return type:MigrationTable instance
Exception:MigrationException
check(name=None)

Get check

Parameters:*columns – List of the column’s name
Return type:MigrationConstraintCheck instance
column(name=None)

Get Column

Parameters:name – Column name
Return type:MigrationColumn instance
drop()

Drop the table

index(*columns, **kwargs)

Get index

Parameters:*columns – List of the column’s name
Return type:MigrationIndex instance
primarykey()

Get primary key

Parameters:*columns – List of the column’s name
Return type:MigrationConstraintPrimaryKey instance
unique(*columns, **kwargs)

Get unique

Parameters:*columns – List of the column’s name
Return type:MigrationConstraintUnique instance
class anyblok.migration.Migration(session, metadata)

Migration Main entry

This class allows to manipulate all the migration class:

migration = Migration(Session(), Base.Metadata)
t = migration.table('My table name')
c = t.column('My column name from t')
auto_upgrade_database()

Upgrade the database automaticly

detect_changed()

Detect the difference between the metadata and the database

Return type:MigrationReport instance
release_savepoint(name)

Release the save point

Parameters:name – name of the savepoint
rollback_savepoint(name)

Rollback to the savepoint

Parameters:name – name of the savepoint
savepoint(name=None)

Add a savepoint

Parameters:name – name of the save point
Return type:return the name of the save point
table(name=None)

Get a table

Return type:MigrationTable instance

anyblok._graphviz module

class anyblok._graphviz.BaseSchema(name, format='png')

Common class extended by the type of schema

add_edge(cls_1, cls_2, attr=None)

Add new edge between 2 node

dot.add_edge(node1, node2)
Parameters:
  • cls_1 – node (string or object) for the from
  • cls_2 – node (string or object) for the to
Paam attr:

attribute of the edge

render()

Call graphviz to do the schema

save()

render and create the output file

class anyblok._graphviz.SQLSchema(name, format='png')

Create a schema to display the table model

dot = SQLSchema('the name of my schema')
t1 = dot.add_table('Table 1')
t1.add_column('c1', 'Integer')
t1.add_column('c2', 'Integer')
t2 = dot.add_table('Table 2')
t2.add_column('c1', 'Integer')
t2.add_foreign_key(t1, 'c2')
dot.save()
add_label(name)

Add a new node TableSchema without column

Parameters:name – name of the table
Return type:return the instance of TableSchema
add_table(name)

Add a new node TableSchema with column

Parameters:name – name of the table
Return type:return the instance of TableSchema
get_table(name)

Return the instance of TableSchema linked with the name of table

Parameters:name – name of the table
Return type:return the instance of TableSchema
class anyblok._graphviz.TableSchema(name, parent, islabel=False)

Describe one table

add_column(name, type_, primary_key=False)

Add a new column in the table

Parameters:
  • name – name of the column
  • type – type of the column
  • primary_key – if True, the string PK will be add
add_foreign_key(node, label=None, nullable=True)

Add a new foreign key

Parameters:
  • node – node (string or object) of the table linked
  • label – name of the column of the foreign key
  • nullable – bool to select the multiplicity of the association
render(dot)

Call graphviz to create the schema

class anyblok._graphviz.ModelSchema(name, format='png')

Create a schema to display the UML model

dot = ModelSchema('The name of my UML schema')
cls = dot.add_class('My class')
cls.add_method('insert')
cls.add_property('items')
cls.add_column('my column')
dot.save()
add_class(name)

Add a new node ClassSchema with column

Parameters:name – name of the class
Return type:return the instance of ClassSchema
add_label(name)

Return the instance of ClassSchema linked with the name of class

Parameters:name – name of the class
Return type:return the instance of ClassSchema
get_class(name)

Add a new node ClassSchema without column

Parameters:name – name of the class
Return type:return the instance of ClassSchema
class anyblok._graphviz.ClassSchema(name, parent, islabel=False)

Use to display a class

add_column(name)

add a column in the class

Parameters:name – name of the column
add_method(name)

add a method in the class

Parameters:name – name of the method
add_property(name)

add a property in the class

Parameters:name – name of the property
agregate(node, label_from=None, multiplicity_from=None, label_to=None, multiplicity_to=None)

add an edge with agregate shape to the node

Parameters:
  • node – node (string or object)
  • label_from – attribute name
  • multiplicity_from – multiplicity of the attribute
  • label_to – attribute name
  • multiplicity_to – multiplicity of the attribute
associate(node, label_from=None, multiplicity_from=None, label_to=None, multiplicity_to=None)

add an edge with associate shape to the node

Parameters:
  • node – node (string or object)
  • label_from – attribute name
  • multiplicity_from – multiplicity of the attribute
  • label_to – attribute name
  • multiplicity_to – multiplicity of the attribute
extend(node)

add an edge with extend shape to the node

Parameters:node – node (string or object)
render(dot)

Call graphviz to do the schema

strong_agregate(node, label_from=None, multiplicity_from=None, label_to=None, multiplicity_to=None)

add an edge with strong agregate shape to the node

Parameters:
  • node – node (string or object)
  • label_from – attribute name
  • multiplicity_from – multiplicity of the attribute
  • label_to – attribute name
  • multiplicity_to – multiplicity of the attribute

anyblok.databases module

Management of the database

adapter = getUtility(ISqlAlchemyDataBase, drivername)
adapter.createdb(dbname)
logger.info(adapter.listdb())
adapter.dropdb(dbname)

anyblok.databases.postgres module

class anyblok.databases.postgres.SqlAlchemyPostgres

Postgres adapter for database management

cnx()

Context manager to get a connection to database

createdb(dbname)

Create a database

Parameters:dbname – database name to create
dropdb(dbname)

Drop a database

Parameters:dbname – database name to drop
listdb()

list database

Return type:list of database name

anyblok.scripts module

anyblok.scripts.createdb(description, argsparse_groups, parts_to_load)

Create a database and install blok from config

Parameters:
  • description – description of argsparse
  • argsparse_groups – list argsparse groupe to load
  • parts_to_load – group of blok to load
anyblok.scripts.updatedb(description, version, argsparse_groups, parts_to_load)

Update an existing database

Parameters:
  • description – description of argsparse
  • version – version of script for argparse
  • argsparse_groups – list argsparse groupe to load
  • parts_to_load – group of blok to load
anyblok.scripts.interpreter(description, version, argsparse_groups, parts_to_load)

Execute a script or open an interpreter

Parameters:
  • description – description of argsparse
  • version – version of script for argparse
  • argsparse_groups – list argsparse groupe to load
  • parts_to_load – group of blok to load
anyblok.scripts.sqlschema(description, version, argsparse_groups, parts_to_load)

Create a Table model schema of the registry

Parameters:
  • description – description of argsparse
  • version – version of script for argparse
  • argsparse_groups – list argsparse groupe to load
  • parts_to_load – group of blok to load
anyblok.scripts.modelschema(description, version, argsparse_groups, parts_to_load)

Create a UML model schema of the registry

Parameters:
  • description – description of argsparse
  • version – version of script for argparse
  • argsparse_groups – list argsparse groupe to load
  • parts_to_load – group of blok to load