Amazon Web Services (AWS) Relational Database Service (RDS) is a powerful and scalable database solution that simplifies the process of setting up, operating, and scaling relational databases in the cloud. In this tutorial, we will explore how to leverage the capabilities of AWS RDS using Python, guiding you through the steps to establish a connection and perform a complete CRUD (Create, Read, Update, Delete) operations.
In this tutorial we assume that you already have an RDS configured with MySQL or MariaDB. You will need the RDS user, password and endpoint. As a next step you will need a server with Python 3 installed. The server can be any operating system, but we will be using Ubuntu.
To connect to mysql, type in the terminal. Replace user and endpoint with your bank’s settings.
mysql -u user -h endpoint -p
After pressing enter, type in your bank password.
CREATE DATABASE IF NOT EXISTS dbClass;
Enter the command to create the table:
USE dbClass;
and:
CREATE TABLE `tbBook` ( `codBook` int(11) NOT NULL, `nameBook` varchar(255) NOT NULL, `yearBook` varchar(10) NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_general_ci;
To display the tables created, type:
SHOW TABLES;
Type exit; to return to the operating system terminal.
We will use our classes folder to store the classes and files for connections to the database. Create the following files in the folder: conexao.py, LivroDAO.py and MenuLivros.py.
from classes.MenuLivros import MenuLivros menu = MenuLivros() menu.executar()
import mysql.connector
def conectar():
mydb = mysql.connector.connect(
host="localhost",
user="root",
password="",
database="dbClass"
)
return mydb
In this file we put the settings for accessing the bank. Replace the data in quotes with the values from your RDS installation.
import mysql.connector
from classes.conexao import conectar
class LivroDAO:
def __init__(self):
self.conexao = conectar()
def listar_livros(self):
try:
cursor = self.conexao.cursor()
cursor.execute("SELECT * FROM tbBook")
livros = cursor.fetchall()
return livros
except mysql.connector.Error as erro:
print(f"Erro ao listar livros: {erro}")
finally:
cursor.close()
def buscar_livro(self, codBook):
try:
cursor = self.conexao.cursor()
cursor.execute("SELECT * FROM tbBook WHERE codBook = %s", (codBook,))
livro = cursor.fetchone()
return livro
except mysql.connector.Error as erro:
print(f"Erro ao buscar livro: {erro}")
finally:
cursor.close()
def inserir_livro(self, nameBook, yearBook):
try:
cursor = self.conexao.cursor()
query = "INSERT INTO tbBook (nameBook, yearBook) VALUES (%s, %s)"
valores = (nameBook, yearBook)
cursor.execute(query, valores)
self.conexao.commit()
print("Livro inserido com sucesso.")
except mysql.connector.Error as erro:
print(f"Erro ao inserir livro: {erro}")
finally:
cursor.close()
def atualizar_livro(self, codBook, nameBook, yearBook):
try:
cursor = self.conexao.cursor()
query = "UPDATE tbBook SET nameBook = %s, yearBook = %s WHERE codBook = %s"
valores = (nameBook, yearBook, codBook)
cursor.execute(query, valores)
self.conexao.commit()
print("Livro atualizado com sucesso.")
except mysql.connector.Error as erro:
print(f"Erro ao atualizar livro: {erro}")
finally:
cursor.close()
def excluir_livro(self, codBook):
try:
cursor = self.conexao.cursor()
query = "DELETE FROM tbBook WHERE codBook = %s"
valores = (codBook,)
cursor.execute(query, valores)
self.conexao.commit()
print("Livro excluído com sucesso.")
except mysql.connector.Error as erro:
print(f"Erro ao excluir livro: {erro}")
finally:
cursor.close()
from classes.LivroDAO import LivroDAO
class MenuLivros:
def __init__(self):
self.dao = LivroDAO()
def exibir_menu(self):
print("--- MENU ---")
print("1. List Books")
print("2. Search Books")
print("3. Insert Books")
print("4. Update Books")
print("5. Delete Books")
print("0. Exit")
def listar_livros(self):
livros = self.dao.listar_livros()
if livros:
print("Books:")
for livro in livros:
print(f"Code: {livro[0]}, Name: {livro[1]}, Year: {livro[2]}")
else:
print("No books found.")
def buscar_livro(self):
codLivro = input("Enter the code of the book you want to search for: ")
livro = self.dao.buscar_livro(codLivro)
if livro:
print(f"Code: {livro[0]}, Name: {livro[1]}, Year: {livro[2]}")
else:
print("Book not found.")
def inserir_livro(self):
nomeLivro = input("Enter the name of the book: ")
anoLivro = input("Enter the year of the book: ")
self.dao.inserir_livro(nomeLivro, anoLivro)
print("Book successfully inserted.")
def atualizar_livro(self):
codLivro = input("Enter the code of the book you want to update: ")
nomeLivro = input("Enter the new name of the book: ")
anoLivro = input("Enter the new year of the book: ")
self.dao.atualizar_livro(codLivro, nomeLivro, anoLivro)
print("Successfully updated book.")
def excluir_livro(self):
codLivro = input("Enter the code of the book you want to delete: ")
self.dao.excluir_livro(codLivro)
print("Book successfully deleted.")
def executar(self):
while True:
self.exibir_menu()
opcao = input("Enter the desired option: ")
if opcao == "1":
self.listar_livros()
elif opcao == "2":
self.buscar_livro()
elif opcao == "3":
self.inserir_livro()
elif opcao == "4":
self.atualizar_livro()
elif opcao == "5":
self.excluir_livro()
elif opcao == "0":
print("Closing the program.")
break
else:
print("Invalid option. Type again.")
Install the library to integrate Python with the Bank.
pip install mysql-connector-python
Run the application.py program.
python application.py
Conclusion
By the end of this tutorial, you’ll have a comprehensive understanding of connecting Python applications to AWS RDS and performing CRUD operations. Whether you are a beginner or an experienced developer, mastering these skills will empower you to build scalable and efficient cloud-based applications with AWS RDS.
Enhance your development journey by embracing the power of AWS RDS and Python, and unlock the full potential of cloud-based relational databases for your projects.