DEV Community

Marcelo Costa
Marcelo Costa

Posted on • Edited on

2

Using Python to create MySQL tables with random schema

Having a large amount of test data sometimes take a lot of effort, and to simulate a more realistic scenario, it’s good to have a large number of tables with distinct column types. This script generates random tables schema for MySQL.

Environment

Activate your virtualenv
pip install --upgrade virtualenv
python3 -m virtualenv --python python3 env
source ./env/bin/activate
Enter fullscreen mode Exit fullscreen mode
Install the requirements for the metadata generator
pip install -r requirements.txt
Enter fullscreen mode Exit fullscreen mode

Code

mysql-connector-python
import argparse
import logging
import random
import sys
import uuid
from mysql.connector import connect
_DATA_TYPES = [
'INT', 'TINYINT', 'SMALLINT', 'MEDIUMINT',
'FLOAT', 'DOUBLE(10,2)', 'DECIMAL(10,2)', 'CHAR(5)', 'TINYTEXT', 'TEXT',
'MEDIUMTEXT', 'MEDIUMBLOB', 'LONGTEXT', 'LONGBLOB', 'DATE',
'TIME', 'BOOL', 'DATETIME'
]
_COLUMN_NAMES = [
'name', 'address', 'city', 'state', 'date_time', 'paragraph', 'randomdata',
'person', 'credit_card', 'size', 'reason', 'school', 'food', 'location',
'house', 'price', 'cpf', 'cnpj', 'passport', 'security_number',
'phone_number', 'bank_account_number', 'ip_address', 'stocks'
]
_DESCRIPTION_VALUES = [
'This is a random generated column',
'Description for random generated column'
]
_TABLE_NAMES = [
'school_info', 'personal_info', 'persons', 'employees', 'companies',
'store', 'home'
]
_SCHEMA_NAMES = [
'school_warehouse', 'company_warehouse', 'on_prem_warehouse',
'factory_warehouse', 'organization_warehouse'
]
def get_conn(connection_args):
return connect(database=connection_args['database'],
host=connection_args['host'],
username=connection_args['user'],
password=connection_args['pass'])
def create_random_metadata(connection_args):
conn = get_conn(connection_args)
cursor = conn.cursor()
for x in range(4):
schema_name, schema_stmt = build_create_schema_statement()
cursor.execute(schema_stmt)
for y in range(250):
query = build_create_table_statement(schema_name)
print('\n' + query)
cursor.execute(query)
conn.commit()
cursor.close()
def get_random_data_type():
return random.choice(_DATA_TYPES)
def get_random_column_name():
return random.choice(_COLUMN_NAMES)
def get_random_column_description():
return random.choice(_DESCRIPTION_VALUES)
def get_random_table_name():
return random.choice(_TABLE_NAMES)
def get_random_schema_name():
return random.choice(_SCHEMA_NAMES)
def build_create_table_statement(schema_name):
table_stmt = 'CREATE TABLE {}.{}{} ( '.format(schema_name,
get_random_table_name(),
uuid.uuid4().hex[:8])
table_stmt = '{}{}{} {}'.format(table_stmt, get_random_column_name(),
str(random.randint(1, 100000)),
get_random_data_type())
for x in range(random.randint(1, 15)):
table_stmt += ', {}{}'.format(get_random_column_name(),
str(random.randint(1, 100000))) + \
' {}'.format(get_random_data_type())
table_stmt = '{} )'.format(table_stmt)
return table_stmt
def build_create_schema_statement():
schema_name = '{}{}'.format(get_random_schema_name(),
str(random.randint(1, 100000)))
schema_stmt = 'CREATE SCHEMA {} '.format(schema_name)
return schema_name, schema_stmt
def parse_args():
parser = argparse.ArgumentParser(
description='Command line generate random metadata into mysql')
parser.add_argument('--mysql_host',
help='Your mysql server host, this is required even'
' for the raw_metadata_csv,'
' so we are able to map the created entries'
' resource with the mysql host',
required=True)
parser.add_argument('--mysql_user', help='Your mysql credentials user')
parser.add_argument('--mysql_pass', help='Your mysql credentials password')
parser.add_argument('--mysql_database', help='Your mysql database name')
return parser.parse_args()
if __name__ == "__main__":
args = parse_args()
# Enable logging
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
create_random_metadata({
'database': args.mysql_database,
'host': args.mysql_host,
'user': args.mysql_user,
'pass': args.mysql_pass
})

Execution

export MYSQL_SERVER=127.0.0.1
export MYSQL_USERNAME=test_user
export MYSQL_PASSWORD=test_user_pwd
export MYSQL_DATABASE=test_db

python metadata_generator.py \
--mysql_host=$MYSQL_SERVER \
--mysql_user=$MYSQL_USERNAME \
--mysql_pass=$MYSQL_PASSWORD \
--mysql_database=$MYSQL_DATABASE
Enter fullscreen mode Exit fullscreen mode

And that's it!

If you have difficulties, don’t hesitate reaching out. I would love to help you!

API Trace View

How I Cut 22.3 Seconds Off an API Call with Sentry 👀

Struggling with slow API calls? Dan Mindru walks through how he used Sentry's new Trace View feature to shave off 22.3 seconds from an API call.

Get a practical walkthrough of how to identify bottlenecks, split tasks into multiple parallel tasks, identify slow AI model calls, and more.

Read more →

Top comments (0)

Billboard image

The Next Generation Developer Platform

Coherence is the first Platform-as-a-Service you can control. Unlike "black-box" platforms that are opinionated about the infra you can deploy, Coherence is powered by CNC, the open-source IaC framework, which offers limitless customization.

Learn more

AWS GenAI Live!

GenAI LIVE! is a dynamic live-streamed show exploring how AWS and our partners are helping organizations unlock real value with generative AI.

Tune in to the full event

DEV is partnering to bring live events to the community. Join us or dismiss this billboard if you're not interested. ❤️