DEV Community

Marcelo Costa
Marcelo Costa

Posted on • Edited on

3 1

Using Python to create SqlServer 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 SQLServer.

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
Install ODBC Driver 17 for SQL Server

https://docs.microsoft.com/en-us/sql/connect/odbc/linux-mac/installing-the-microsoft-odbc-driver-for-sql-server?view=sql-server-2017

Code

pyodbc
import argparse
import logging
import random
import sys
import uuid
from pyodbc import connect
_DATA_TYPES = [
'INT', 'TINYINT', 'SMALLINT',
'FLOAT', 'REAL', 'CHAR(5)', 'VARCHAR(25)', 'TEXT',
'BINARY', 'DATE', 'TIME', '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'
]
_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 build_connection(database, host, user, password):
return 'DRIVER={ODBC Driver 17 for SQL Server};' + \
'SERVER={};DATABASE={};UID={};PWD={}'.format(
host, database, user, password)
def get_conn(connection_args):
return connect(build_connection(database=connection_args['database'],
host=connection_args['host'],
user=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_schema_name():
return random.choice(_SCHEMA_NAMES)
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 get_random_data_type():
return random.choice(_DATA_TYPES)
def get_random_column_name():
return random.choice(_COLUMN_NAMES)
def get_random_table_name():
return random.choice(_TABLE_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 parse_args():
parser = argparse.ArgumentParser(
description='Command line generate random metadata into sqlserver')
parser.add_argument('--sqlserver-host',
help='Your sqlserver server host',
required=True)
parser.add_argument('--sqlserver-user', help='Your sqlserver credentials user')
parser.add_argument('--sqlserver-pass', help='Your sqlserver credentials password')
parser.add_argument('--sqlserver-database', help='Your sqlserver 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.sqlserver_database,
'host': args.sqlserver_host,
'user': args.sqlserver_user,
'pass': args.sqlserver_pass
})

Execution

export SQLSERVER_SERVER=127.0.0.1
export SQLSERVER_USERNAME=sqlserver
export SQLSERVER_PASSWORD=sqlserver_pwd
export SQLSERVER_DATABASE=test_db

python metadata_generator.py \
--sqlserver-host=$SQLSERVER_SERVER \ 
--sqlserver-user=$SQLSERVER_USERNAME \
--sqlserver-pass=$SQLSERVER_PASSWORD \ 
--sqlserver-database=$SQLSERVER_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!

Sentry image

Hands-on debugging session: instrument, monitor, and fix

Join Lazar for a hands-on session where you’ll build it, break it, debug it, and fix it. You’ll set up Sentry, track errors, use Session Replay and Tracing, and leverage some good ol’ AI to find and fix issues fast.

RSVP here →

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

👋 Kindness is contagious

Please leave a ❤️ or a friendly comment on this post if you found it helpful!

Okay