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] 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))
[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)
: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.encode()
return dump_data
[docs] def restore_db(self, password, dbname, data, **kwargs):
""" Restore database
: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()