aiida.utils package

Submodules

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.format_call_graph(calc_node, info_fn=<function calc_info>)[source]

Print a tree like the POSIX tree command for the calculation call graph

Parameters:
  • calc_node – The calculation node
  • info_fn – An optional function that takes the node and returns a string of information to be displayed for each node.
class aiida.utils.capturing.Capturing(capture_stderr=False)[source]

Bases: object

This class captures stdout and returns it (as a list, split by lines).

Note: if you raise a SystemExit, you have to catch it outside. E.g., in our tests, this works:

import sys
with self.assertRaises(SystemExit):
    with Capturing() as output:
        sys.exit()

But out of the testing environment, the code instead just exits.

To use it, access the obj.stdout_lines, or just iterate over the object

Parameters:capture_stderr – if True, also captures sys.stderr. To access the lines, use obj.stderr_lines. If False, obj.stderr_lines is None.
__dict__ = dict_proxy({'__module__': 'aiida.utils.capturing', '__exit__': <function __exit__>, '__dict__': <attribute '__dict__' of 'Capturing' objects>, '__str__': <function __str__>, '__enter__': <function __enter__>, '__weakref__': <attribute '__weakref__' of 'Capturing' objects>, '__iter__': <function __iter__>, '__init__': <function __init__>, '__doc__': '\n This class captures stdout and returns it\n (as a list, split by lines).\n\n Note: if you raise a SystemExit, you have to catch it outside.\n E.g., in our tests, this works::\n\n import sys\n with self.assertRaises(SystemExit):\n with Capturing() as output:\n sys.exit()\n\n But out of the testing environment, the code instead just exits.\n\n To use it, access the obj.stdout_lines, or just iterate over the object\n\n :param capture_stderr: if True, also captures sys.stderr. To access the\n lines, use obj.stderr_lines. If False, obj.stderr_lines is None.\n '})
__enter__()[source]
__exit__(*args)[source]
__init__(capture_stderr=False)[source]

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

__iter__()[source]
__module__ = 'aiida.utils.capturing'
__str__() <==> str(x)[source]
__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>, '_backend': <property object>, '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)

_backend

Get the backend

_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]
aiida.utils.which.which(cmd, mode=1, path=None)[source]

Given a command, mode, and a PATH string, return the path which conforms to the given mode on the PATH, or None if there is no such file.

mode defaults to os.F_OK | os.X_OK. path defaults to the result of os.environ.get(“PATH”), or can be overridden with a custom search path.