假设你要开发一个新的数据库解决方案，需要真实的数据和用户账号进行测试，并且一条命令完成该任务。实现过程已说明了，来看看完整的代码copy_server.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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
#
# Copyright (c) 2010, 2013, Oracle and/or its affiliates. All rights reserved.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; version 2 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
#
“”“
This file contains an example of how to build a customized utility using
the MySQL Utilities scripts and libraries.
““”
import optparse
import os
import sys
from mysql.utilities import VERSION_FRM
from mysql.utilities.command import dbcopy
from mysql.utilities.command import serverclone
from mysql.utilities.command import userclone
from mysql.utilities.common.server import Server
from mysql.utilities.common.options import parse_connection
from mysql.utilities.exception import UtilError
# Constants
NAME = “example – copy_server “
DESCRIPTION = “copy_server – copy an existing server”
USAGE = “%prog –server=user:pass@host:port:socket “ \
“–new-dir=<path> –new-id=<server_id> “ \
“–new-port=<port> –databases=<db list> “ \
“–users=<user list>”
# Setup the command parser
parser = optparse.OptionParser(
version=VERSION_FRM.format(program=os.path.basename(sys.argv[0])),
description=DESCRIPTION,
usage=USAGE,
add_help_option=False)
parser.add_option(“–help”, action=“help”)
# Setup utility-specific options:
# Connection information for the source server
parser.add_option(“–server”, action=“store”, dest=“server”,
type=“string”, default=“root@localhost:3306”,
help=“connection information for original server in “ + \
“the form: <user>:<password>@<host>:<port>:<socket>”)
# Data directory for new instance
parser.add_option(“–new-data”, action=“store”, dest=“new_data”,
type=“string”, help=“the full path to the location “
“of the data directory for the new instance”)
# Port for the new instance
parser.add_option(“–new-port”, action=“store”, dest=“new_port”,
type=“string”, default=“3307”, help=“the new port “
“for the new instance – default=%default”)
# Server id for the new instance
parser.add_option(“–new-id”, action=“store”, dest=“new_id”,
type=“string”, default=“2”, help=“the server_id for “
“the new instance – default=%default”)
# List of databases
parser.add_option(“-d”, “–databases”, action=“store”, dest=“dbs_to_copy”,
type=“string”, help=“comma-separated list of databases “
“to include in the copy (omit for all databases)”,
default=None)
# List of users
parser.add_option(“-u”, “–users”, action=“store”, dest=“users_to_copy”,
type=“string”, help=“comma-separated list of users “
“to include in the copy (omit for all users)”,
default=None)
# Now we process the rest of the arguments.
opt, args = parser.parse_args()
# Parse source connection values
try:
conn = parse_connection(opt.server)
except:
parser.error(“Server connection values invalid or cannot be parsed.”)
# Get a server class instance
print “# Connecting to server…”
server_options = {
‘conn_info’ : conn,
‘role’ : “source”,
}
server1 = Server(server_options)
try:
server1.connect()
except UtilError, e:
print “ERROR:”, e.errmsg
# Get list of databases from the server if not specified in options
print “# Getting databases…”
db_list = []
if opt.dbs_to_copy is None:
for db in server1.get_all_databases():
db_list.append((db[0], None))
else:
for db in opt.dbs_to_copy.split(“,”):
db_list.append((db, None))
# Get list of all users from the server
print “# Getting users…”
user_list=[]
if opt.users_to_copy is None:
users = server1.exec_query(“SELECT user, host “
“FROM mysql.user “
“WHERE user != ‘root’ and user != ””)
for user in users:
user_list.append(user[0]+‘@’+user[1])
else:
for user in opt.users_to_copy.split(“,”):
user_list.append(user)
# Build options
options = {
‘new_data’ : opt.new_data,
‘new_port’ : opt.new_port,
‘new_id’ : opt.new_id,
‘root_pass’ : ‘root’,
‘mysqld_options’ : ‘–report-host=localhost –report-port=%s’ % opt.new_port,
}
# Clone the server
print “# Cloning server instance…”
try:
res = serverclone.clone_server(conn, options)
except UtilError, e:
print “ERROR:”, e.errmsg
sys.exit()
# Set connection values
dest_values = {
“user” : conn.get(“user”),
“passwd” : “root”,
“host” : conn.get(“host”),
“port” : opt.new_port,
“unix_socket” : os.path.join(opt.new_data, “mysql.sock”)
}
# Build dictionary of options
options = {
“quiet” : True,
“force” : True
}
print “# Copying databases…”
try:
dbcopy.copy_db(conn, dest_values, db_list, options)
except UtilError, e:
print “ERROR:”, e.errmsg
sys.exit()
# Build dictionary of options
options = {
“overwrite” : True,
“quiet” : True,
“globals” : True
}
print “# Cloning the users…”
for user in user_list:
try:
res = userclone.clone_user(conn, dest_values, user,
(user,), options)
except UtilError, e:
print “ERROR:”, e.errmsg
sys.exit()
print “# …done.”
|
该代码需要Python基础，有兴趣的去学习下Python。
文章转载来自：ttlsa.com
