In this entry, we will explore how to use Python to connect to an Oracle MySQL database and invoke a stored procedure. This is particularly useful when you're looking to interact with your database programmatically—whether it's for inserting data, querying, or managing business logic through stored procedures.
We will walk through a Python script that connects to the database, generates some random test data, in this case telemetry data for school busses, and invokes a stored procedure to insert the data into multiple tables. Below is the full Python code, and we’ll break down each part of the code in detail to help you understand how it works.
This assumes that you have Oracle MySQL Running, a stored procedure to call, and the proper credentials to access your database, you can modify this script and use a stored procedure of your own!
Here's the full Python script that connects to the MySQL database, generates random test data, and calls the stored procedure InsertEvents.
Script:
import mysql.connector
from mysql.connector import errorcode
import random
import datetime
# Database connection parameters
config = {
'user': 'yourusername',
'password': 'yourpassword',
'host': 'ip address of vm running mysql',
'port': port you are using,
'database': 'name of your mysql database',
}
# Function to generate random test data
def generate_test_data():
asset_id = random.randint(1000, 9999)
happened_at_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
latitude = round(random.uniform(-90, 90), 6)
longitude = round(random.uniform(-180, 180), 6)
heading_degrees = random.randint(0, 360)
accuracy_meters = round(random.uniform(0.5, 20.0), 2)
geofence = 'TestGeofence' + str(random.randint(1, 100))
gps_speed = round(random.uniform(0, 30), 2)
ecu_speed = round(random.uniform(0, 30), 2)
return (asset_id, happened_at_time, latitude, longitude, heading_degrees, accuracy_meters, geofence, gps_speed, ecu_speed)
# Function to insert event data by calling the stored procedure InsertEvents
def insert_event_data(cursor, data):
try:
cursor.callproc('InsertEvents', data)
print(f"Successfully inserted event for asset_id {data[0]}")
except mysql.connector.Error as err:
print(f"Error: {err}")
return False
return True
# Main function to connect to the database and insert events
def main():
try:
# Connect to the database
cnx = mysql.connector.connect(**config)
cursor = cnx.cursor()
# Ask the user how many events to generate
num_events = int(input("Enter the number of events to generate: "))
# Generate and insert the specified number of test events
for _ in range(num_events):
test_data = generate_test_data()
if insert_event_data(cursor, test_data):
cnx.commit()
else:
cnx.rollback()
cursor.close()
cnx.close()
except mysql.connector.Error as err:
if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
print("Something is wrong with your username and/or password")
elif err.errno == errorcode.ER_BAD_DB_ERROR:
print("Database does not exist or is not reachable")
else:
print(err)
if __name__ == "__main__":
main()
