# -*- 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.entities.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