odoo_17.0.1/core/web/controllers/database.py

179 lines
8.2 KiB
Python
Raw Normal View History

# Part of Odoo. See LICENSE file for full copyright and licensing details.
import datetime
import logging
import os
import re
import tempfile
from lxml import html
import odoo
import odoo.modules.registry
from odoo import http
from odoo.http import content_disposition, dispatch_rpc, request, Response
from odoo.service import db
from odoo.tools.misc import file_open, str2bool
from odoo.tools.translate import _
from odoo.addons.base.models.ir_qweb import render as qweb_render
_logger = logging.getLogger(__name__)
DBNAME_PATTERN = '^[a-zA-Z0-9][a-zA-Z0-9_.-]+$'
class Database(http.Controller):
def _render_template(self, **d):
d.setdefault('manage', True)
d['insecure'] = odoo.tools.config.verify_admin_password('admin')
d['list_db'] = odoo.tools.config['list_db']
d['langs'] = odoo.service.db.exp_list_lang()
d['countries'] = odoo.service.db.exp_list_countries()
d['pattern'] = DBNAME_PATTERN
# databases list
try:
d['databases'] = http.db_list()
d['incompatible_databases'] = odoo.service.db.list_db_incompatible(d['databases'])
except odoo.exceptions.AccessDenied:
d['databases'] = [request.db] if request.db else []
templates = {}
with file_open("web/static/src/public/database_manager.qweb.html", "r") as fd:
templates['database_manager'] = fd.read()
with file_open("web/static/src/public/database_manager.master_input.qweb.html", "r") as fd:
templates['master_input'] = fd.read()
with file_open("web/static/src/public/database_manager.create_form.qweb.html", "r") as fd:
templates['create_form'] = fd.read()
def load(template_name):
fromstring = html.document_fromstring if template_name == 'database_manager' else html.fragment_fromstring
return (fromstring(templates[template_name]), template_name)
return qweb_render('database_manager', d, load)
@http.route('/web/database/selector', type='http', auth="none")
def selector(self, **kw):
if request.db:
request.env.cr.close()
return self._render_template(manage=False)
@http.route('/web/database/manager', type='http', auth="none")
def manager(self, **kw):
if request.db:
request.env.cr.close()
return self._render_template()
@http.route('/web/database/create', type='http', auth="none", methods=['POST'], csrf=False)
def create(self, master_pwd, name, lang, password, **post):
insecure = odoo.tools.config.verify_admin_password('admin')
if insecure and master_pwd:
dispatch_rpc('db', 'change_admin_password', ["admin", master_pwd])
try:
if not re.match(DBNAME_PATTERN, name):
raise Exception(_('Houston, we have a database naming issue! Make sure you only use letters, numbers, underscores, hyphens, or dots in the database name, and you\'ll be golden.'))
# country code could be = "False" which is actually True in python
country_code = post.get('country_code') or False
dispatch_rpc('db', 'create_database', [master_pwd, name, bool(post.get('demo')), lang, password, post['login'], country_code, post['phone']])
request.session.authenticate(name, post['login'], password)
request.session.db = name
return request.redirect('/web')
except Exception as e:
_logger.exception("Database creation error.")
error = "Database creation error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
@http.route('/web/database/duplicate', type='http', auth="none", methods=['POST'], csrf=False)
def duplicate(self, master_pwd, name, new_name, neutralize_database=False):
insecure = odoo.tools.config.verify_admin_password('admin')
if insecure and master_pwd:
dispatch_rpc('db', 'change_admin_password', ["admin", master_pwd])
try:
if not re.match(DBNAME_PATTERN, new_name):
raise Exception(_('Houston, we have a database naming issue! Make sure you only use letters, numbers, underscores, hyphens, or dots in the database name, and you\'ll be golden.'))
dispatch_rpc('db', 'duplicate_database', [master_pwd, name, new_name, neutralize_database])
if request.db == name:
request.env.cr.close() # duplicating a database leads to an unusable cursor
return request.redirect('/web/database/manager')
except Exception as e:
_logger.exception("Database duplication error.")
error = "Database duplication error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
@http.route('/web/database/drop', type='http', auth="none", methods=['POST'], csrf=False)
def drop(self, master_pwd, name):
insecure = odoo.tools.config.verify_admin_password('admin')
if insecure and master_pwd:
dispatch_rpc('db', 'change_admin_password', ["admin", master_pwd])
try:
dispatch_rpc('db', 'drop', [master_pwd, name])
if request.session.db == name:
request.session.logout()
return request.redirect('/web/database/manager')
except Exception as e:
_logger.exception("Database deletion error.")
error = "Database deletion error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
@http.route('/web/database/backup', type='http', auth="none", methods=['POST'], csrf=False)
def backup(self, master_pwd, name, backup_format='zip'):
insecure = odoo.tools.config.verify_admin_password('admin')
if insecure and master_pwd:
dispatch_rpc('db', 'change_admin_password', ["admin", master_pwd])
try:
odoo.service.db.check_super(master_pwd)
ts = datetime.datetime.utcnow().strftime("%Y-%m-%d_%H-%M-%S")
filename = "%s_%s.%s" % (name, ts, backup_format)
headers = [
('Content-Type', 'application/octet-stream; charset=binary'),
('Content-Disposition', content_disposition(filename)),
]
dump_stream = odoo.service.db.dump_db(name, None, backup_format)
response = Response(dump_stream, headers=headers, direct_passthrough=True)
return response
except Exception as e:
_logger.exception('Database.backup')
error = "Database backup error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
@http.route('/web/database/restore', type='http', auth="none", methods=['POST'], csrf=False, max_content_length=None)
def restore(self, master_pwd, backup_file, name, copy=False, neutralize_database=False):
insecure = odoo.tools.config.verify_admin_password('admin')
if insecure and master_pwd:
dispatch_rpc('db', 'change_admin_password', ["admin", master_pwd])
try:
data_file = None
db.check_super(master_pwd)
with tempfile.NamedTemporaryFile(delete=False) as data_file:
backup_file.save(data_file)
db.restore_db(name, data_file.name, str2bool(copy), neutralize_database)
return request.redirect('/web/database/manager')
except Exception as e:
error = "Database restore error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
finally:
if data_file:
os.unlink(data_file.name)
@http.route('/web/database/change_password', type='http', auth="none", methods=['POST'], csrf=False)
def change_password(self, master_pwd, master_pwd_new):
try:
dispatch_rpc('db', 'change_admin_password', [master_pwd, master_pwd_new])
return request.redirect('/web/database/manager')
except Exception as e:
error = "Master password update error: %s" % (str(e) or repr(e))
return self._render_template(error=error)
@http.route('/web/database/list', type='json', auth='none')
def list(self):
"""
Used by Mobile application for listing database
:return: List of databases
:rtype: list
"""
return http.db_list()