aiida.utils package

Submodules

aiida.utils.ascii_vis._ctime(node)[source]
aiida.utils.ascii_vis._generate_node_label(node, node_attr, show_pk)[source]

Generate a label for the node.

Parameters:
  • node (aiida.orm.node.Node) – The node to generate the label for
  • node_attr (str) – The attribute to use as the label, can be None
  • show_pk (bool) – if True, show the PK alongside the label
Returns:

The generated label

Return type:

str

aiida.utils.ascii_vis.build_tree(node, node_label=None, show_pk=True, max_depth=1, follow_links_of_type=None, descend=True, depth=0)[source]
aiida.utils.ascii_vis.draw_children(node, node_label=None, show_pk=True, dist=2, follow_links_of_type=None)[source]

Print an ASCII tree of the parents of the given node.

Parameters:
  • node (aiida.orm.data.Data) – The node to draw for
  • node_label (str) – The label to use for the nodes
  • show_pk (bool) – Show the PK of nodes alongside the label
  • dist (int) – The number of steps away from this node to branch out
  • follow_links_of_type (str) – Follow links of this type when making steps, if None then it will follow CREATE and INPUT links
aiida.utils.ascii_vis.draw_parents(node, node_label=None, show_pk=True, dist=2, follow_links_of_type=None)[source]

Print an ASCII tree of the parents of the given node.

Parameters:
  • node (aiida.orm.data.Data) – The node to draw for
  • node_label (str) – The label to use for the nodes
  • show_pk (bool) – Show the PK of nodes alongside the label
  • dist (int) – The number of steps away from this node to branch out
  • follow_links_of_type (str) – Follow links of this type when making steps, if None then it will follow CREATE and INPUT links
aiida.utils.ascii_vis.get_ascii_tree(node, node_label=None, show_pk=True, max_depth=1, follow_links_of_type=None, descend=True)[source]

Get a string representing an ASCII tree for the given node.

Parameters:
  • node (aiida.orm.node.Node) – The node to get the tree for
  • node_label (str) – What to label the nodes with (can be an attribute name)
  • show_pk (bool) – If True, show the pk with the node label
  • max_depth (int) – The maximum depth to follow starting from the node
  • follow_links_of_type (One of the members from aiida.common.links.LinkType) – Follow links of a given type, can be None
  • descend (bool) – if True will follow outputs, if False inputs
Returns:

The string giving an ASCII representation of the tree from the node

Return type:

str

aiida.utils.calculation.add_source_info(node, func)[source]

Take information about a function using inspection and add it as attributes of a node.

Parameters:
  • node – The node to add the information to.
  • func – The function to inspect.
class aiida.utils.capturing.Capturing[source]

Bases: list

__dict__ = dict_proxy({'__module__': 'aiida.utils.capturing', '__exit__': <function __exit__>, '__enter__': <function __enter__>, '__dict__': <attribute '__dict__' of 'Capturing' objects>, '__weakref__': <attribute '__weakref__' of 'Capturing' objects>, '__doc__': None})
__enter__()[source]
__exit__(*args)[source]
__module__ = 'aiida.utils.capturing'
__weakref__

list of weak references to the object (if defined)

aiida.utils.delete_nodes.delete_nodes(pks, follow_calls=False, follow_returns=False, dry_run=False, force=False, disable_checks=False, verbosity=0)[source]

Delete nodes by a list of pks

Note:

The script will also delete all children calculations generated from the specified nodes.

Parameters:
  • pks – a list of the PKs of the nodes to delete
  • follow_calls (bool) – Follow calls
  • follow_returns (bool) – Follow returns. This is a very dangerous option, since anything returned by a workflow might have been used as input in many other calculations. Use with care, and never combine with force.
  • dry_run (bool) – Do not delete, a dry run, with statistics printed according to verbosity levels.
  • force (bool) – Do not ask for confirmation to delete nodes.
  • disable_checks (bool) – If True, will not check whether calculations are losing created data or called instances. If checks are disabled, also logging is disabled.
  • force – Do not ask for confirmation to delete nodes.
  • verbosity (int) – The verbosity levels, 0 prints nothing, 1 prints just sums and total, 2 prints individual nodes.
aiida.utils.email.normalize_email(email)[source]

Normalize the address by lowercasing the domain part of the email address.

Taken from Django.

aiida.utils.find_folder.find_path(root, dir_name)[source]

Testing tools for related projects like plugins.

Fixtures (pytest) and specific test classes & test runners (unittest) that set up a complete temporary AiiDA environment for plugin tests.

Filesystem:

  • temporary config (.aiida) folder
  • temporary repository folder

Database:

  • temporary database cluster via the pgtest package
  • with aiida database user
  • with aiida_db database

AiiDA:

  • set to use the temporary config folder
  • create and configure a profile
exception aiida.utils.fixtures.FixtureError(msg)[source]

Bases: exceptions.Exception

Raised by FixtureManager, when it encounters a situation in which consistent behaviour can not be guaranteed

__init__(msg)[source]

x.__init__(…) initializes x; see help(type(x)) for signature

__module__ = 'aiida.utils.fixtures'
__str__() <==> str(x)[source]
__weakref__

list of weak references to the object (if defined)

class aiida.utils.fixtures.FixtureManager[source]

Bases: object

Manage the life cycle of a completely separated and temporary AiiDA environment

  • No previously created database of profile is required to run tests using this environment
  • Tests using this environment will never pollute the user’s work environment

Example:

fixtures = FixtureManager()
fixtures.create_aiida_db()  # set up only the database
fixtures.create_profile()  # set up a profile (creates the db too if necessary)

# ready for testing

# run test 1

fixtures.reset_db()
# database ready for independent test 2

# run test 2

fixtures.destroy_all()
# everything cleaned up

Usage (unittest): See the PluginTestCase and the TestRunner.

Usage (pytest):

import pytest

@pytest.fixture(scope='session')
def aiida_profile():
    with aiida.utils.fixtures.fixture_manager() as fixture_mgr:
        fixture_mgr.create_profile()
        yield fixture_manager

@pytest.fixture(scope='function')
def test_data(aiida_profile):
    # load my test data
    yield
    fixture_manager.reset_db()

def test_my_stuff(test_data):
    # run a test
static _FixtureManager__clean_db_django()
_FixtureManager__clean_db_sqla()

Clean database for sqlalchemy backend

__dict__ = dict_proxy({'profile': <property object>, '__module__': 'aiida.utils.fixtures', 'create_aiida_db': <function create_aiida_db>, 'root_dir': <property object>, '__dict__': <attribute '__dict__' of 'FixtureManager' objects>, 'config_dir_ok': <property object>, 'create_db_cluster': <function create_db_cluster>, 'repo': <property object>, 'db_pass': <property object>, 'last_name': <property object>, 'create_profile': <function create_profile>, 'db_port': <property object>, 'config_dir': <property object>, 'institution': <property object>, '__init__': <function __init__>, 'backend': <property object>, 'first_name': <property object>, 'has_profile_open': <function has_profile_open>, 'db_user': <property object>, '__weakref__': <attribute '__weakref__' of 'FixtureManager' objects>, 'create_root_dir': <function create_root_dir>, '__doc__': "\n Manage the life cycle of a completely separated and temporary AiiDA environment\n\n * No previously created database of profile is required to run tests using this\n environment\n * Tests using this environment will never pollute the user's work environment\n\n Example::\n\n fixtures = FixtureManager()\n fixtures.create_aiida_db() # set up only the database\n fixtures.create_profile() # set up a profile (creates the db too if necessary)\n\n # ready for testing\n\n # run test 1\n\n fixtures.reset_db()\n # database ready for independent test 2\n\n # run test 2\n\n fixtures.destroy_all()\n # everything cleaned up\n\n Usage (unittest): See the :py:class:`PluginTestCase` and the :py:class:`TestRunner`.\n\n\n Usage (pytest)::\n\n import pytest\n\n @pytest.fixture(scope='session')\n def aiida_profile():\n with aiida.utils.fixtures.fixture_manager() as fixture_mgr:\n fixture_mgr.create_profile()\n yield fixture_manager\n\n @pytest.fixture(scope='function')\n def test_data(aiida_profile):\n # load my test data\n yield\n fixture_manager.reset_db()\n\n def test_my_stuff(test_data):\n # run a test\n\n\n ", '_FixtureManager__clean_db_django': <staticmethod object>, 'db_name': <property object>, 'db_host': <property object>, 'destroy_all': <function destroy_all>, 'reset_db': <function reset_db>, 'root_dir_ok': <property object>, '_FixtureManager__clean_db_sqla': <function __clean_db_sqla>, '_return_dir': <function _return_dir>, 'email': <property object>, 'repo_ok': <function repo_ok>})
__init__()[source]

x.__init__(…) initializes x; see help(type(x)) for signature

__module__ = 'aiida.utils.fixtures'
__weakref__

list of weak references to the object (if defined)

_return_dir(key)[source]

Return a path to a directory from the fs environment

backend
config_dir
config_dir_ok
create_aiida_db()[source]

Create the necessary database on the temporary postgres instance

create_db_cluster()[source]
create_profile()[source]

Set AiiDA to use the test config dir and create a default profile there

Warning: the AiiDA dbenv must not be loaded when this is called!

create_root_dir()[source]
db_host
db_name
db_pass
db_port
db_user
destroy_all()[source]

Remove all traces of the test run

email
first_name
has_profile_open()[source]
institution
last_name
profile

Profile parameters

repo
repo_ok()[source]
reset_db()[source]

Cleans all data from the database between tests

root_dir
root_dir_ok
class aiida.utils.fixtures.PluginTestCase(methodName='runTest')[source]

Bases: unittest.case.TestCase

Set up a complete temporary AiiDA environment for plugin tests.

Note: This test class needs to be run through the TestRunner and will not work simply with python -m unittest discover.

Usage example:

MyTestCase(aiida.utils.fixtures.PluginTestCase):

    def setUp(self):
        # load my test data

    # optionally extend setUpClass / tearDownClass / tearDown if needed

    def test_my_plugin(self):
        # execute tests
__module__ = 'aiida.utils.fixtures'
classmethod setUpClass()[source]

Hook method for setting up class fixture before running tests in the class.

tearDown()[source]

Hook method for deconstructing the test fixture after testing it.

class aiida.utils.fixtures.TestRunner(stream=<open file '<stderr>', mode 'w'>, descriptions=True, verbosity=1, failfast=False, buffer=False, resultclass=None)[source]

Bases: unittest.runner.TextTestRunner

Testrunner for unit tests using the fixture manager.

Usage example:

import unittest
from aiida.utils.fixtures import TestRunner

tests = unittest.defaultTestLoader.discover('.')
TestRunner().run(tests)
__module__ = 'aiida.utils.fixtures'
run(suite, backend='django')[source]

Run tests using fixture manager for specified backend.

Parameters:
  • tests – A suite of tests, as returned e.g. by unittest.TestLoader.discover()
  • backend – Database backend to be used.
aiida.utils.fixtures.fixture_manager(*args, **kwds)[source]

Context manager for FixtureManager objects

Example test runner (unittest):

with fixture_manager() as fixture_mgr:
    # ready for tests
# everything cleaned up

Example fixture (pytest):

def aiida_profile():
    with fixture_manager() as fixture_mgr:
        yield fixture_mgr
aiida.utils.queries._get_column(colname, alias)[source]

Return the column for the projection, if the column name is specified.

class aiida.utils.queries.array_length(*clauses, **kwargs)[source]

Bases: sqlalchemy.sql.functions.FunctionElement

__module__ = 'aiida.utils.queries'
_compiler_dispatch(**kw)
_compiler_dispatcher = <sqlalchemy.ext.compiler._dispatcher object>
name = 'array_len'
aiida.utils.queries.compile(element, compiler, **kw)[source]

Get length of array defined in a JSONB column

class aiida.utils.queries.jsonb_array_length(*clauses, **kwargs)[source]

Bases: sqlalchemy.sql.functions.FunctionElement

__module__ = 'aiida.utils.queries'
_compiler_dispatch(**kw)
_compiler_dispatcher = <sqlalchemy.ext.compiler._dispatcher object>
name = 'jsonb_array_len'
class aiida.utils.queries.jsonb_typeof(*clauses, **kwargs)[source]

Bases: sqlalchemy.sql.functions.FunctionElement

__module__ = 'aiida.utils.queries'
_compiler_dispatch(**kw)
_compiler_dispatcher = <sqlalchemy.ext.compiler._dispatcher object>
name = 'jsonb_typeof'
aiida.utils.serialize.decode_key(key)[source]

Helper function for the deserialize_data function which can undo the key encoding of tuple keys done by the encode_key function

Parameters:key – the key to decode
Returns:the decoded key
aiida.utils.serialize.deserialize_data(data)[source]

Deserialize a single value or a collection that may contain serialized AiiDA nodes. This is essentially the inverse operation of serialize_data which will reload node instances from the serialized UUID data. Encoded tuples that are used as dictionary keys will be decoded.

Parameters:data – serialized data
Returns:the deserialized data with keys decoded and node instances loaded from UUID’s
aiida.utils.serialize.encode_key(key)[source]

Helper function for the serialize_data function which may need to serialize a dictionary that uses tuples as keys. This function will encode the tuple into a string such that it is JSON serializable

Parameters:key – the key to encode
Returns:the encoded key
aiida.utils.serialize.serialize_data(data)[source]

Serialize a value or collection that may potentially contain AiiDA nodes, which will be serialized to their UUID. Keys encountered in any mappings, such as a dictionary, will also be encoded if necessary. An example is where tuples are used as keys in the pseudo potential input dictionaries. These operations will ensure that the returned data is JSON serializable.

Parameters:data – a single value or collection
Returns:the serialized data with the same internal structure
class aiida.utils.test_timezone.TimezoneTest(methodName='runTest')[source]

Bases: unittest.case.TestCase

__module__ = 'aiida.utils.test_timezone'
test_timezone_now()[source]
aiida.utils.timezone.delta(from_time, to_time=None)[source]
aiida.utils.timezone.get_current_timezone()[source]
aiida.utils.timezone.is_aware(value)[source]
aiida.utils.timezone.is_naive(value)[source]
aiida.utils.timezone.localtime(value, timezone=None)[source]

Converts an aware datetime.datetime to local time. Local time is defined by the current time zone, unless another time zone is specified.

aiida.utils.timezone.make_aware(value, timezone=None, is_dst=None)[source]
Parameters:
  • value (datetime.datetime) – The datetime to make aware
  • timezone
  • is_dst
Returns:

aiida.utils.timezone.now()[source]