Source code for aiida.orm.implementation.sql.backends

# -*- coding: utf-8 -*-
###########################################################################
# Copyright (c), The AiiDA team. All rights reserved.                     #
# This file is part of the AiiDA code.                                    #
#                                                                         #
# The code is hosted on GitHub at https://github.com/aiidateam/aiida-core #
# For further information on the license, see the LICENSE.txt file        #
# For further information please visit http://www.aiida.net               #
###########################################################################
"""Generic backend related objects"""

import abc
import typing

from .. import backends

__all__ = ('SqlBackend',)

# The template type for the base ORM model type
ModelType = typing.TypeVar('ModelType')  # pylint: disable=invalid-name


[docs]class SqlBackend(typing.Generic[ModelType], backends.Backend): """ A class for SQL based backends. Assumptions are that: * there is an ORM * that it is possible to convert from ORM model instances to backend instances * that psycopg2 is used as the engine if any of these assumptions do not fit then just implement a backend from :class:`aiida.orm.implementation.Backend` """
[docs] @abc.abstractmethod def get_backend_entity(self, model): """ Return the backend entity that corresponds to the given Model instance :param model: the ORM model instance to promote to a backend instance :return: the backend entity corresponding to the given model :rtype: :class:`aiida.orm.implementation.BackendEntity` """
[docs] @abc.abstractmethod def cursor(self): """ Return a psycopg cursor. This method should be used as a context manager i.e.:: with backend.cursor(): # Do stuff :return: a psycopg cursor :rtype: :class:`psycopg2.extensions.cursor` """
[docs] @abc.abstractmethod def execute_raw(self, query): """Execute a raw SQL statement and return the result. :param query: a string containing a raw SQL statement :return: the result of the query """
[docs] def execute_prepared_statement(self, sql, parameters): """Execute an SQL statement with optional prepared statements. :param sql: the SQL statement string :param parameters: dictionary to use to populate the prepared statement """ results = [] with self.cursor() as cursor: cursor.execute(sql, parameters) for row in cursor: results.append(row) return results