-
Notifications
You must be signed in to change notification settings - Fork 58
/
copydb.py
135 lines (123 loc) · 4.41 KB
/
copydb.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
#!/usr/bin/env python
# Copyright 2018 ACSONE SA/NV (<http://acsone.eu>)
# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html).
import os
import shutil
import subprocess
import click
import click_odoo
from click_odoo import odoo
from psycopg2.extensions import AsIs, quote_ident
from ._dbutils import (
db_exists,
pg_connect,
reset_config_parameters,
terminate_connections,
)
def _copy_db(cr, source, dest):
cr.execute(
"CREATE DATABASE %s WITH TEMPLATE %s",
(AsIs(quote_ident(dest, cr)), AsIs(quote_ident(source, cr))),
)
def _copy_filestore(source, dest, copy_mode="default"):
filestore_source = odoo.tools.config.filestore(source)
if os.path.isdir(filestore_source):
filestore_dest = odoo.tools.config.filestore(dest)
if copy_mode == "hardlink" or copy_mode == "rsync":
try:
if copy_mode == "hardlink":
hardlink_option = ["--link-dest=" + filestore_source]
else:
hardlink_option = []
cmd = (
[
"rsync",
"-a",
"--delete-delay",
]
+ hardlink_option
+ [
filestore_source + "/",
filestore_dest,
]
)
subprocess.check_call(cmd)
# we use one generic exception clause here because subprocess.check_call
# may not only raise the documented subprocess.CalledProcessError
# (when the command exits with a return code != 0) but also with at least a
# few other Exceptions like PermissionError when the given command is not
# executable (by the current user) or a FileNotFoundError if the given
# command is not in the users PATH or cannot be found on the system
except Exception as e:
msg = "Error syncing filestore to: {}, {}".format(dest, e)
raise click.ClickException(msg)
else:
shutil.copytree(filestore_source, filestore_dest)
@click.command()
@click_odoo.env_options(
default_log_level="warn", with_database=False, with_rollback=False
)
@click.option(
"--force-disconnect",
"-f",
is_flag=True,
help="Attempt to disconnect users from the template database.",
)
@click.option(
"--unless-dest-exists",
is_flag=True,
help="Don't report error if destination database already exists.",
)
@click.option(
"--if-source-exists",
is_flag=True,
help="Don't report error if source database does not exist.",
)
@click.option(
"--filestore-copy-mode",
type=click.Choice(["default", "rsync", "hardlink"]),
default="default",
help="Mode for copying the filestore. Default uses python shutil copytree "
"which copies everything. If the target filestore already exists and "
"just needs an update you can use rsync to rsync the filestore "
"instead. If both the target filestore already exists and is on the same "
"disk you might use hardlink which hardlinks all files to the inode in the "
"source filestore and saves you space.",
)
@click.argument("source", required=True)
@click.argument("dest", required=True)
def main(
env,
source,
dest,
force_disconnect,
unless_dest_exists,
if_source_exists,
filestore_copy_mode,
):
"""Create an Odoo database by copying an existing one.
This script copies using postgres CREATEDB WITH TEMPLATE.
It also copies the filestore.
"""
with pg_connect() as cr:
if db_exists(dest):
msg = "Destination database already exists: {}".format(dest)
if unless_dest_exists:
click.echo(click.style(msg, fg="yellow"))
return
else:
raise click.ClickException(msg)
if not db_exists(source):
msg = "Source database does not exist: {}".format(source)
if if_source_exists:
click.echo(click.style(msg, fg="yellow"))
return
else:
raise click.ClickException(msg)
if force_disconnect:
terminate_connections(source)
_copy_db(cr, source, dest)
reset_config_parameters(dest)
_copy_filestore(source, dest, copy_mode=filestore_copy_mode)
if __name__ == "__main__": # pragma: no cover
main()