Instructobit
In this tutorial we will be using the official python MySQL connector package to make our connections and execute our queries:
pip install mysql-connector-python
This package is also available directly from the MySQL documentation site. For a more detailed explanation and quick overview of the mysql connector package.
Making the connection
The first thing we'll need to do of course is make the connection to our database, we'll be using this connection for all the following examples in this tutorial, if you are unfamiliar with the MySQL connector, the official Python MySQL connector documentation has some pretty good examples.
The you can change the credentials in the below example can be changed to your correct credentials, we'll be using a dictionary cursor for our executions just to make the selection results easier to read.
import mysql.connector
host_args = {
"host": "localhost",
"user": "testuser",
"password": "testpass"
}
con = mysql.connector.connect(**host_args)
cur = con.cursor(dictionary=True)
Setting up a test .sql file with multiple statements for execution
Now we need to create a file containing the multiple statements we wish to execute, for this first example we'll create a simple test database and table and insert a couple of records into that table
test1.sql
DROP DATABASE IF EXISTS sql_test;
CREATE DATABASE sql_test;
USE sql_test;
CREATE TABLE test_table(
id INT PRIMARY KEY AUTO_INCREMENT,
test_field VARCHAR(250) NOT NULL DEFAULT ''
);
INSERT INTO test_table(test_field) VALUES('test1');
INSERT INTO test_table(test_field) VALUES('test2');
INSERT INTO test_table(test_field) VALUES('test3');
If you're running a Linux distributions like Ubuntu you could execute the above file from the command line like this:
root@TEST-PC:~# mysql -h localhost -u testuser -ptestpass < test1.sql
Executing the entire .sql file within a single cursor object in Python
To execute the entire file, we must first open it and read the contents into the cursor.execute() method, however this would not work without the additional use of the 'multi=True' argument which allows for multiple statements to be provided to the method.
When the multi argument is provided, the execute statement will return an iterator that when looped over will return the cursor object at the time of each statements execution, each statement within the file will be executed sequentially. For each iteration we can access the cursor object and print information about the query it is currently executing.
# Your connection code here...
with open('test1.sql', 'r') as sql_file:
result_iterator = cur.execute(sql_file.read(), multi=True)
for res in result_iterator:
print("Running query: ", res) # Will print out a short representation of the query
print(f"Affected {res.rowcount} rows" )
con.commit() # Remember to commit all your changes!
Running the above example code will output the following:
Running query: CMySQLCursorDict: DROP DATABASE IF EXISTS sql_test
Affected 1 rows
Running query: CMySQLCursorDict: CREATE DATABASE sql_test
Affected 1 rows
Running query: CMySQLCursorDict: USE sql_test
Affected 0 rows
Running query: CMySQLCursorDict: CREATE TABLE test_table(
id INT ..
Affected 0 rows
Running query: CMySQLCursorDict: INSERT INTO test_table(test_field) VALUE..
Affected 1 rows
Running query: CMySQLCursorDict: INSERT INTO test_table(test_field) VALUE..
Affected 1 rows
Running query: CMySQLCursorDict: INSERT INTO test_table(test_field) VALUE..
Affected 1 rows
If you now access your database and have a look in the newly created table you should see all 3 inserts:
mysql> select * from sql_test.test_table;
+----+----------------+
| id | test_field |
+----+----------------+
| 1 | test1 |
| 2 | test variables |
| 3 | test3 |
+----+----------------+
Handling any potential output from selects and other output-yielding statements
If you include any SQL statements in your file that yield some form of result, most likely from a SELECT statement or SHOW statement for example, you'll need to handle the result set when iterating over it to continue with the iteration, to know whether or not output rows should be handled, you can use the cursor.with_rows attribute which will return True if there are rows to fetch. Then you can handle them as you see fit, for this example we will simply print out the result.
For this example we will using the following SQL file:
test2.sql
USE sql_test;
SELECT * FROM test_table;
SHOW VARIABLES LIKE 'sql_mode';
Here we are simply selecting the data from the test table we created earlier, then listing the current value of our sql_mode system variable.
To read the results of this file we can edit our code slightly:
import json # Just using this to give a prettier result
# Your connection code here...
with open('test2.sql', 'r') as sql_file:
result_iterator = cur.execute(sql_file.read(), multi=True)
for res in result_iterator:
print("Running query: ", res)
if res.with_rows:
fetch_result = res.fetchall()
print(json.dumps(fetch_result, indent=4))
elif res.rowcount > 0:
print(f"Affected {res.rowcount} rows" )
con.commit()
Running this code will result in the following output:
Running query: CMySQLCursorDict: USE sql_test
Running query: CMySQLCursorDict: SELECT * FROM test_table
[
{
"id": 1,
"test_field": "test1"
},
{
"id": 2,
"test_field": "test2"
},
{
"id": 3,
"test_field": "test3"
}
]
Running query: CMySQLCursorDict: SHOW VARIABLES LIKE 'sql_mode'
[
{
"Variable_name": "sql_mode",
"Value": "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION"
}
]
Setting and using user and system variables
Without any modification to our code we can also make use of user and system set variables as well as updating them within your session, allowing for more complexity within your .sql files
test3.sql
USE sql_test;
SET @test_var = 'test variables';
INSERT INTO test_table(test_field) VALUES(@test_var);
SELECT * FROM test_table;
SHOW VARIABLES LIKE 'sql_mode';
SET SESSION sql_mode='NO_ZERO_IN_DATE,NO_ZERO_DATE';
SHOW VARIABLES LIKE 'sql_mode';
Just update which .sql file your Python code points to then execute it, the result will be the following:
Running query: CMySQLCursorDict: USE sql_test
Running query: CMySQLCursorDict: SET @test_var = 'test variables'
Running query: CMySQLCursorDict: INSERT INTO test_table(test_field) VALUE..
Affected 1 rows
Running query: CMySQLCursorDict: SELECT * FROM test_table
[
{
"id": 1,
"test_field": "test1"
},
{
"id": 2,
"test_field": "test2"
},
{
"id": 3,
"test_field": "test3"
},
{
"id": 4,
"test_field": "test variables"
}
]
Running query: CMySQLCursorDict: SHOW VARIABLES LIKE 'sql_mode'
[
{
"Variable_name": "sql_mode",
"Value": "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION"
}
]
Running query: CMySQLCursorDict: SET SESSION sql_mode='NO_ZERO_IN_DATE,NO..
Running query: CMySQLCursorDict: SHOW VARIABLES LIKE 'sql_mode'
[
{
"Variable_name": "sql_mode",
"Value": "NO_ZERO_IN_DATE,NO_ZERO_DATE"
}
]
This way of executing .sql files can be extremely powerful and allows you to move complex SQL queries out of your python code and into independent files, allowing for better organization for large projects with many complex database operations.
Executing .sql files in Python with string parameters
If you want to avoid SQL injection the easiest way to protect yourself using Python MySQL is to use string parameters, these are cleaned before use within the statement. Executing an ordinary SQL statement with parameters in Python looks like this:
sql = "INSERT INTO test_table(test_field) VALUES(%(Param)s);"
params = {
'Param': "testparam"
}
cur.execute(sql_file.read(), params)
We can add parameters to our SQL file just as easily. Simply add the parameter string into your SQL like so:
test4.sql
INSERT INTO test_table(test_field) VALUES('test1');
INSERT INTO test_table(test_field) VALUES(%(Param)s);
INSERT INTO test_table(test_field) VALUES('test3');
SELECT * FROM test_table;
We can then assign a value to our parameter in the Python code:
with open('test4.sql', 'r') as sql_file:
params = {
'Param': "testparam"
}
cur_result = cur.execute(sql_file.read(), params, multi=True)
for res in cur_result:
print("Running query: ", res)
if res.with_rows:
fetch_result = res.fetchall()
print(json.dumps(fetch_result, indent=4))
elif res.rowcount > 0:
print(f"Affected {res.rowcount} rows" )
con.commit()
This will result in the following output:
Running query: CMySQLCursorDict: INSERT INTO test_table(test_field) VALUE..
Affected 1 rows
Running query: CMySQLCursorDict: INSERT INTO test_table(test_field) VALUE..
Affected 1 rows
Running query: CMySQLCursorDict: INSERT INTO test_table(test_field) VALUE..
Affected 1 rows
Running query: CMySQLCursorDict: SELECT * FROM test_table
[
{
"id": 1,
"test_field": "test1"
},
{
"id": 2,
"test_field": "testparam"
},
{
"id": 3,
"test_field": "test3"
}
]
If you have any questions or would like to add anything interesting, please feel free to leave a comment below!
