This is a tutorial on how to create a CRUD operation service with Python that can handle MySQL data. Note that a prerequisite for this tutorial is that you have Python installed. You can follow this guide in order to learn how to set it up.
Prerequisites
Project Structure
Setting up MySQL
Follow this tutorial in order to setup the MySQL environment on your computer. You will be ready in no time!
Initializing Pipenv & Installing Packages
Before you can initialize Pipenv, you obviously need to install it. If you haven't done so yet, I have a small tutorial on how to set it up here.
In your project root, run the following command to initialize Pipenv (our virtual environment).
Terminal
pipenv shellThis will create a Pipfile in our root folder which will hold information about what packages we've installed.
Next we need to install MySQL Connector Python Package
Terminal
pipenv install mysql-connector-pythonThen, since we're using pipenv we'll need to select the correct Python interpreter. If you're using VSCode you can set it by pressing CMD + SHIFT + P and typing "interpreter" in the command prompt that opens up.
Press ENTER and a list should appear. Select the interpreter that has your project folder name in its name, as well as pipenv. So based on the image below, choose the interpreter at the bottom.
Creating the Database
Time to setup our database and tables. Start off by creating a new file called database.py in your projects root folder.
database.py
import mysql.connector as connector
config = {
'user': 'root',
'password': 'password123',
'host': 'localhost'
}
db = connector.connect(**config)
cursor = db.cursor()First we want to connect to our database. So let's start by importing the connector module from the mysql package and naming it connector.
Then we need to create a config dictionary with the credentials needed to connect to our database. Note that you'll have to replace password123 with your own MySQL password.
Next, we're creating a new variable called db which will hold the connection instance. We create this instance by calling the connect method of the connector. Note that since we're passing a dictionary as an argument to the connect method and that method only accepts keyword arguments we'll have to use the double asterisk ** before passing it which will convert the dictionary to keyword arguments.
Then we'll create a cursor which will be used to execute queries.
Next we'll create a new file called setup.py in our project root folder.
setup.py
import mysql.connector as connector
from database import cursor
DB_NAME = 'mydb'
def create_database():
cursor.execute("CREATE DATABASE IF NOT EXISTS {} DEFAULT CHARACTER SET 'utf8'".format(DB_NAME))
print("Database {} created!".format(DB_NAME))
create_database()Just like before we import the connector. We also import the cursor that we created before, from database.py.
Now let's name the database that we're about to create. We store the preferred name in a variable DB_NAME.
Next, we define a function create_database which will use the cursor to execute an SQL statement that creates the database with the DB_NAME that we previously defined. Note that within the SQL statement we're using string formatting, meaning that we can append the value of DB_NAME within the string at the location of the square brackets {}.
Lastly, for our function, we print out a message which notifies us about the database creation. We then call the function.
Now let's try this out and see if it works! So in our terminal that's running our virtual environment (pipenv) we can now run setup.py.
Terminal
python setup.pyNow you should see the print message that we defined in the create_database function, displayed in the terminal. If not, then be sure to double check that you typed in the correct credentials before connecting to the database in database.py.
We can also verify that our database was created by checking it in MySQL Workbench that we installed earlier. Note that you might have to right click the schema panel and click Refresh All if you do not see it.
Creating the Tables
Now, that we have created the database, the time has come for us to setup the tables.
We'll start by importing (below the previous imports) errorcode from the connector module in setup.py. This is something that we will use later.
setup.py
from mysql.connector import errorcodeBelow the imports of setup.py we'll input the following:
setup.py
TABLES = {}
TABLES['users'] = (
"CREATE TABLE `users` ("
" `id` int(11) NOT NULL AUTO_INCREMENT,"
" `username` varchar(255) NOT NULL,"
" `password` varchar(255) NOT NULL,"
" `created` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,"
"PRIMARY KEY (`id`)"
") ENGINE=InnoDB"
)We create an empty TABLES dictionary which we'll add SQL statements to.
We define our first table users which we create by defining it as a key on the empty TABLES dictionary. We then set the value to an SQL statement for creating a table called users which will consist of four columns; id, username, password and created. More on SQL queries and the meaning of the query will come in another tutorial.
Now it's time to create a new function create_tables. You can create it below the create_database function definition that we previously created.
setup.py
def create_tables():
cursor.execute("USE {}".format(DB_NAME))
for table_name in TABLES:
table_description = TABLES[table_name]
try:
print("Creating table ({})".format(table_name))
cursor.execute(table_description)
except connector.Error as err:
if err.errno == errorcode.ER_TABLE_EXISTS_ERROR:
print("Table ({}) already exists".format(table_name))
else:
print(err.msg)Here we need to first use the cursor to execute a SQL statement where we're telling MySQL that we want to use the database that we previously created.
We then use a for loop to loop through the TABLES dictionary. On each iteration we're creating a variable table_description which will hold the value of the current index of TABLES.
Next we'll define try/except block where we try to execute the SQL statement for the current table, using the cursors execute method.
If this operation fails for any reason it will cast an MySQL exception which we're handling in the except block. We can use the error objects errno field (error number) in a logical expression, comparing it with one of the errorcode object's (that we imported earlier) fields to handle that specific error code. In this example we're checking for the error code that's being thrown when we're trying to create a table that already exists. If we get any other error code, we handle that by printing the error message in the else block.
This is what your code in setup.py should look like by now.
setup.py
import mysql.connector as connector
from mysql.connector import errorcode
from database import cursor
DB_NAME = 'mydb'
TABLES = {}
TABLES['users'] = (
"CREATE TABLE `users` ("
" `id` int(11) NOT NULL AUTO_INCREMENT,"
" `username` varchar(255) NOT NULL,"
" `password` varchar(255) NOT NULL,"
" `created` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,"
"PRIMARY KEY (`id`)"
") ENGINE=InnoDB"
)
def create_database():
cursor.execute("CREATE DATABASE IF NOT EXISTS {} DEFAULT CHARACTER SET 'utf8'".format(DB_NAME))
print("Database {} created!".format(DB_NAME))
def create_tables():
cursor.execute("USE {}".format(DB_NAME))
for table_name in TABLES:
table_description = TABLES[table_name]
try:
print("Creating table ({})".format(table_name))
cursor.execute(table_description)
except connector.Error as err:
if err.errno == errorcode.ER_TABLE_EXISTS_ERROR:
print("Table ({}) already exists".format(table_name))
else:
print(err.msg)
create_database()
create_tables()Now let's see if this works!
Terminal
python setup.pyWhen running this the first time, we should see the print message we defined earlier saying "Creating table (yourdatabasename)". If we run this command once again we should get notified that the table already exists.
We can also confirm that the tables were created in MySQL Workbench.
Creating Functions for CRUD Operations
Now it's time to create some functions for handling CRUD operations, but before we do that we want to let our database connection know which database to connect to while performing these CRUD operations. So in our database.py, we will want to add the name of the database, that we previously created, as a key/value pair in the config dictionary. Note that this should be the same value as the DB_NAME value that we defined in setup.py.
database.py
1import mysql.connector as connector
2
3config = {
4 'user': 'root',
5 'password': 'password123',
6 'host': 'localhost',
7 'database': 'mydb'
8}
9
10db = connector.connect(**config)
11cursor = db.cursor()Note
Note that if you drop (delete) your database at some point in the future and try to create it again as explained in previous examples, you'll have to revert this change since we cannot use a database that has not yet been created.
Now with that change, it's time to create the file main.py where we'll define all our CRUD functions. Start by importing the cursor and the db (database instance).
main.py
from database import cursor, dbPOST Request: Creating users
Below the imports in main.py you should input the following:
main.py
def create_user(username, password):
sql = ("INSERT INTO users(username, password) VALUES (%s, %s)")
cursor.execute(sql, (username, password,))
db.commit()
user_id = cursor.lastrowid
print("Added user {}".format(user_id))
create_user('Christopher', 'pAsSwOrD')
create_user('John', 'doe123')
create_user('Jane', 'hello123')We define a create_user function which will be responsible of creating new users in our database. This function takes two String arguments; username and password.
In our newly created function, we create a variable sql which will hold the MySQL query for adding a new user to the database. The syntax is pretty much self explanatory but for the VALUES part, we define placeholders %s which will be replaced with the arguments that we pass to the cursor's execute method. So we pass username and password in a tuple, as a second argument of the execute method.
Then we need commit this to the database, using the commit method of the db instance. We then get the last commits id and store it in a user_id variable that we lastly print to the console.
Once this is done, we'll call the create_user function 3 times with a dummy username and password to test if we're able to write this data to the database.
Now it's time to run this script and see if the users are added to the database.
Terminal
python main.pyOnce we run this, we should see the print output in the terminal saying that the 3 users where added. Now let's confirm this by double checking in MySQL Workbench.
In MySQL Workbench, right click on users and click Select Rows as shown in the image below.
You should now be able to see that the rows where successfully added to the users table with the correct column values for username and password.
GET Requests: Getting users
Getting all users
Now let's define a second function in the same file main.py right below the create_user function.
main.py
def get_users():
sql = ("SELECT * FROM users ORDER BY created DESC")
cursor.execute(sql)
results = cursor.fetchall()
for row in results:
print(row)We name the function get_users and it's a quite simple function which will select all the users from the database with an SQL query that selects all the users and order them by the created column in descending order.
We store this query as usual in a variable called sql.
Next we execute this query with our cursors execute method and store the results in a variable results by using the cursors fetchall method which returns all the matching rows based on our query.
Lastly we loop through the results and print each row to the console.
Now, after the previous function calls create_user, we can now call get_users. We can also comment out the create_user function calls, since we don't really need to create the users in the database again. So your main.py should now look like this by now:
main.py
from database import cursor, db
def create_user(username, password):
sql = ("INSERT INTO users(username, password) VALUES (%s, %s)")
cursor.execute(sql, (username, password,))
db.commit()
user_id = cursor.lastrowid
print("Added user {}".format(user_id))
def get_users():
sql = ("SELECT * FROM users ORDER BY created DESC")
cursor.execute(sql)
results = cursor.fetchall()
for row in results:
print(row)
# create_user('Christopher', 'pAsSwOrD')
# create_user('John', 'doe123')
# create_user('Jane', 'hello123')
get_users()Now save the changes and run the script.
Terminal
python main.pyYou should now be able to see the users printed to the console.
Getting one user
Now let's create our third CRUD function which will fetch not all users, but one user. Place this new function right below the get_users function.
main.py
def get_user(id):
sql = ("SELECT * FROM users WHERE id = %s")
cursor.execute(sql, (id,))
result = cursor.fetchone()
print(result)We name this function get_user and this function will take an argument which is the id of the user that we want to fetch. So for this sql statement we define that we want all the users where the id is equal to the id that we pass to our function. So in the sql query string, we'll define a placeholder %s and then in the execute method of the cursor object, we pass a touple as a second argument where id is the first index. Note that we need to leave a "hanging comma" after inserting the id since the second argument will be interpreted as a String by Python if we wouldn't do so - and we don't want that, we want Python to interpret this as a touple as the execute method accepts a touple as a second argument.
We define a result again, but this time we're using the fetchone method of the cursor object. We then print the result.
Now let's try to get the user with the id 1, by calling the get_user function and passing 1 (integer) as an argument. We can also comment out the function call that fetches all the users. So your code should look like this so far:
main.py
from database import cursor, db
def create_user(username, password):
sql = ("INSERT INTO users(username, password) VALUES (%s, %s)")
cursor.execute(sql, (username, password,))
db.commit()
user_id = cursor.lastrowid
print("Added user {}".format(user_id))
def get_users():
sql = ("SELECT * FROM users ORDER BY created DESC")
cursor.execute(sql)
results = cursor.fetchall()
for row in results:
print(row)
def get_user(id):
sql = ("SELECT * FROM users WHERE id = %s")
cursor.execute(sql, (id,))
result = cursor.fetchone()
print(result)
# create_user('Christopher', 'pAsSwOrD')
# create_user('John', 'doe123')
# create_user('Jane', 'hello123')
# get_users()
get_user(1)Now save the changes and run the script.
Terminal
python main.pyYou should now be able to see the user with the defined id, printed to the console.
PUT Request: Updating a user
In order to update a user, we'll need to define a new function right below the get_user function. Let's call it update_user_password.
main.py
def update_user_password(id, new_password):
sql = ("UPDATE users SET password = %s WHERE id = %s")
cursor.execute(sql, (new_password, id,))
db.commit()
print("Updated password for user with id {}".format(id))Start by defining the function. This function will take 2 arguments; the id of the user that we want to update the password for as well as the new_password.
In the same manner that you should now be familiar with, we create a sql query with placeholders that get their values from the execute methods second argument where we pass new_password and id as a touple.
We then call the function and comment out previous function calls, meaning your code should look like this:
main.py
from database import cursor, db
def create_user(username, password):
sql = ("INSERT INTO users(username, password) VALUES (%s, %s)")
cursor.execute(sql, (username, password,))
db.commit()
user_id = cursor.lastrowid
print("Added user {}".format(user_id))
def get_users():
sql = ("SELECT * FROM users ORDER BY created DESC")
cursor.execute(sql)
results = cursor.fetchall()
for row in results:
print(row)
def get_user(id):
sql = ("SELECT * FROM users WHERE id = %s")
cursor.execute(sql, (id,))
result = cursor.fetchone()
print(result)
def update_user_password(id, new_password):
sql = ("UPDATE users SET password = %s WHERE id = %s")
cursor.execute(sql, (new_password, id,))
db.commit()
print("Updated password for user with id {}".format(id))
# create_user('Christopher', 'pAsSwOrD')
# create_user('John', 'doe123')
# create_user('Jane', 'hello123')
# get_users()
# get_user(1)
update_user_password(1, 'myNeWpAsSwOrD')Now let's run this.
Terminal
python main.pyAnd we should see the printed message on the console. We can also verify that the password was changed by looking in MySQL Workbench.
DELETE Request: Deleting a user
Now we've finally landed at the last part. Great job so far! So what we want to do now is to create a function which can delete users from the database. We'll input it right below the update_user_password function. Probably no need for me to explain what's going on here since you're familiar with the syntax by now. Just one note, don't forget the "hanging comma" ;)
main.py
def delete_user(id):
sql = ("DELETE FROM users WHERE id = %s")
cursor.execute(sql, (id,))
db.commit()
print("Deleted user with id {}".format(id))Now we want to delete the user and then get all users to see that one user was actually removed. So let's call the delete_user and pass id 1, then we call the get_users function. This means that your code should look like this by now:
main.py
from database import cursor, db
def create_user(username, password):
sql = ("INSERT INTO users(username, password) VALUES (%s, %s)")
cursor.execute(sql, (username, password,))
db.commit()
user_id = cursor.lastrowid
print("Added user {}".format(user_id))
def get_users():
sql = ("SELECT * FROM users ORDER BY created DESC")
cursor.execute(sql)
results = cursor.fetchall()
for row in results:
print(row)
def get_user(id):
sql = ("SELECT * FROM users WHERE id = %s")
cursor.execute(sql, (id,))
result = cursor.fetchone()
print(result)
def update_user_password(id, new_password):
sql = ("UPDATE users SET password = %s WHERE id = %s")
cursor.execute(sql, (new_password, id,))
db.commit()
print("Updated password for user with id {}".format(id))
def delete_user(id):
sql = ("DELETE FROM users WHERE id = %s")
cursor.execute(sql, (id,))
db.commit()
print("Deleted user with id {}".format(id))
# create_user('Christopher', 'pAsSwOrD')
# create_user('John', 'doe123')
# create_user('Jane', 'hello123')
# get_users()
# get_user(1)
# update_user_password(1, 'myNeWpAsSwOrD')
delete_user(1)
get_users()Save the changes and run.
Terminal
python main.pyNow we should see the print message about a user being deleted, followed by 2 rows (the remaining 2 of the previous total of 3) - the remaining users.
As usual, let's also verify that user with id 1 was deleted. Note that you'll need to refresh the list of users in MySQL Workbench as the rows will not update in realtime.
