---
slug: "database-table-copy-from-production-to-devserver-via-python"
title: "Reading DB Settings from Django Configuration and Copying Data from Production to Staging with mysqldump"
description: "I usually write data copy scripts in bash, but I tried writing one in Python this time and decided to document it."
url: "https://www.ytyng.com/en/blog/database-table-copy-from-production-to-devserver-via-python"
publish_date: "2022-11-18T10:16:57Z"
created: "2022-11-18T10:16:57Z"
updated: "2026-02-27T08:49:13.496Z"
categories: ["Django"]
keywords: ""
featured_image_url: "https://media.ytyng.com/resize/20250711/1b2373942036494fa479ae9c9260d637.png.webp?width=768"
has_video: false
has_music: false
video_urls: []
music_urls: []
lang: "en"
---

# Reading DB Settings from Django Configuration and Copying Data from Production to Staging with mysqldump

Copying Data from Production to Development Using `mysqldump` from Django Settings
```python
"""
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)
```