Source code for odoo_rpc_client.service.db

# -*- coding: utf-8 -*-
# Copyright © 2014-2018 Dmytro Katyukha <dmytro.katyukha@gmail.com>

#######################################################################
# This Source Code Form is subject to the terms of the Mozilla Public #
# License, v. 2.0. If a copy of the MPL was not distributed with this #
# file, You can obtain one at http://mozilla.org/MPL/2.0/.            #
#######################################################################

import six
import re
from pkg_resources import parse_version

from ..service.service import ServiceBase

__all__ = ('DBService',)


def to_dbname(db):
    """ Converts db to string, that represents database name

        :param db: Client instance or string with name of database
        :type db: Client|str
        :return: string with database name
        :rtype: str
        :raises ValueError: value of db is not parsable
    """
    from odoo_rpc_client.client import Client
    if isinstance(db, six.string_types):
        return db
    elif isinstance(db, Client) and db.dbname is not None:
        return db.dbname
    else:
        raise ValueError("Wrong value for db!")


[docs]class DBService(ServiceBase): """ Service class to simplify interaction with 'db' service """
[docs] class Meta: name = 'db'
[docs] def list_db(self): """ Display list of databses of thist connection """ return self._service.list()
[docs] def db_exist(self, db): """ Check if database exists :param str|Client db: name of database or *Client* instance with *client.dbname is not None* :return: True if database exists else False :rtype: bool """ return self._service.db_exist(to_dbname(db))
# TODO: add `user` kwarg
[docs] def create_db(self, password, dbname, demo=False, lang='en_US', admin_password='admin'): """ Create new database on server, named *dbname* :param str password: super admin password :param str dbname: name of database to create :param bool demo: load demo data or not. Default: False :param str lang: language to be used for database. Default: 'en_US' :param str admin_password: password to be used for 'Administrator' database user. Default: 'admin' :return: Client instance logged to created database as admin user. :rtype: instance of *odoo_rpc_client.client.Client* """ from odoo_rpc_client.client import Client self.create_database(password, dbname, demo, lang, admin_password) client = Client(self.client.host, port=self.client.port, protocol=self.client.protocol, dbname=dbname, user='admin', pwd=admin_password) return client
[docs] def drop_db(self, password, db): """ Drop specified database :param str password: super admin password :param str|Client db: name of database or *Client* instance with *client.dbname is not None* :raise: `ValueError` (unsupported value of *db* argument) """ return self.drop(password, to_dbname(db))
[docs] def dump_db(self, password, db, **kwargs): """ Dump database Note, that from defined arguments, may be passed other arguments (for example odoo version 9.0 requires format arg to be passed) Note, this method may consume huge amout of memory. In production dump/restore have to be done by other means. :param str password: super admin password :param str|Client db: name of database or *Client* instance with *client.dbname is not None* :param str format: (only odoo 9.0) (default: zip) :raise: `ValueError` (unsupported value of *db* argument) :return: byte-string with base64 encoded data :rtype: bytes """ # format argument available only for odoo version 9.0 if self.server_base_version() >= parse_version('9.0'): args = [kwargs.get('format', 'zip')] else: args = [] dump_data = self.dump(password, to_dbname(db), *args) # Ensure returned data is not unicode, but bytes if isinstance(dump_data, six.text_type): dump_data = dump_data return dump_data
[docs] def restore_db(self, password, dbname, data, **kwargs): """ Restore database Note, this method may consume huge amout of memory. In production dump/restore have to be done by other means. :param str password: super admin password :param str dbname: name of database :param bytes data: restore data (base64 encoded string) :param bool copy: (only odoo 8.0+) if set to True, then new db-uid will be generated. (default: False) :return: True :rtype: bool """ assert isinstance(data, bytes), \ "data must be instance of bytes. got: %s" % type(data) if self.server_base_version() >= parse_version('8.0') and \ 'copy' in kwargs: args = [kwargs['copy']] else: args = [] # data passed to restore should be Unicode, otherwise it dumps it as # binary data on python3 if six.PY3 and isinstance(data, six.binary_type): data = data.decode() return self.restore(password, dbname, data, *args)
[docs] def server_version(self): """ Returns server version. (Already parsed with pkg_resources.parse_version) """ return parse_version(self.server_version_str())
[docs] def server_base_version(self): """ Returns server base version ('9.0', '8.0', etc) parsed via pkg_resources.parse_version. No info about comunity / enterprise here """ base_version = self.server_version().base_version # Remove 'rc.*' suffix base_version = re.sub(r'rc.+', '', base_version) return parse_version(base_version)
[docs] def server_version_str(self): """ Return server version (not wrapped by pkg.parse_version) """ return self._service.server_version()
# make able to check if there some databases on server exists def __contains__(self, name): return name in self.list_db()