Reading DB Settings from Django Configuration and Copying Data from Production to Staging with mysqldump
2022-11-18 19:16 (2 years ago)
ytyng
Copying Data from Production to Development Using mysqldump from Django Settings
"""
Script to copy data from production to dev
This code is not part of Django
"""
import re
import sys
import pathlib
import subprocess
settings_path = pathlib.Path(__file__).parent.parent \
/ 'myproject' / 'myapp' / 'settings'
re_databases = re.compile(r'\nDATABASES = ({[\s\S]+?})\n\n')
def get_database_settings(settings_path: pathlib.Path):
"""
Retrieve DATABASES settings from Django settings file
"""
with open(settings_path) as fp:
content = fp.read()
db_content = re_databases.search(content).group(1)
return eval(db_content)
def dump_and_import(
table_name: str, *, write_to_file: bool = False
) -> tuple[bytes, bytes]:
"""
Dump and import a table
:param table_name: Name of the table
:param write_to_file: If True, write to file and finish. For verification
:return: Standard output, standard error output
"""
dump_command = [
'mysqldump',
'--host=' + production_settings['default']['HOST'],
'--user=' + production_settings['default']['USER'],
'--password=' + production_settings['default']['PASSWORD'],
'--set-gtid-purged=OFF',
'--no-tablespaces',
'--skip-column-statistics',
production_settings['default']['NAME'],
table_name
]
dump_process = subprocess.Popen(
dump_command, stdout=subprocess.PIPE)
if write_to_file:
dump_result = dump_process.communicate()[0]
out_file_path = f'/tmp/{table_name}table_name.mysqldump'
with open(out_file_path, 'wb') as fp:
fp.write(dump_result)
return out_file_path.encode('utf-8') + b' written.', b''
# Convert if the target server does not support utf8mb4_0900_ai_ci
sed_process = subprocess.Popen([
'sed',
's/utf8mb4_0900_ai_ci/utf8mb4_general_ci/g',
], stdout=subprocess.PIPE, stdin=dump_process.stdout)
import_command = [
'mysql',
'--host=' + dev_settings['default']['HOST'],
'--user=' + dev_settings['default']['USER'],
'--password=' + dev_settings['default']['PASSWORD'],
'--database=' + dev_settings['default']['NAME'],
]
import_process = subprocess.Popen(
import_command, stdout=subprocess.PIPE, stdin=sed_process.stdout)
return import_process.communicate()
def drop_table(table_name: str) -> tuple[bytes, bytes]:
drop_command = [
'mysql',
'--host=' + dev_settings['default']['HOST'],
'--user=' + dev_settings['default']['USER'],
'--password=' + dev_settings['default']['PASSWORD'],
'--database=' + dev_settings['default']['NAME'],
f'--execute=DROP TABLE IF EXISTS {table_name}',
]
return subprocess.Popen(drop_command, stdout=subprocess.PIPE).communicate()
table_names_for_copy = [
'myapp_staff',
'myapp_product',
]
if __name__ == '__main__':
dev_settings = get_database_settings(settings_path / 'devserver.py')
production_settings = get_database_settings(settings_path / 'production.py')
for table_name in reversed(table_names_for_copy):
stdout, stderr = drop_table(table_name)
print(stdout.decode('utf-8'))
if stderr:
print(stderr.decode('utf-8'), file=sys.stderr)
for table_name in table_names_for_copy:
print(table_name)
stdout, stderr = dump_and_import(table_name)
print(stdout.decode('utf-8'))
if stderr:
print(stderr.decode('utf-8'), file=sys.stderr)
Comments