Django 設定から DB設定を読み取って、mysqldump で 本番→検証にデータコピーをする
投稿者: ytyng 1年, 5ヶ月 前
Django 設定から DB設定を読み取って、mysqldump で 本番→検証にデータコピーをする
"""
本番 → dev にデータコピーをするスクリプト
このコードは 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):
"""
Django の設定ファイルから DATABASES = の設定を取得する
"""
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]:
"""
テーブルをダンプしてインポートする
:param table_name: テーブル名
:param write_to_file: True にすると、ファイルを書き出して終わる。検証用
:return: 標準出力、標準エラー出力
"""
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''
# import 先サーバが 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)
コメント