Hello again!
On this blog post, I am going to talk about my latest project that I have worked on using Python and SQLAlchemy ver 1.4 with Alembic for database handling.
What is SQLALchemy?
It is a SQL toolkit and object-relational mapper(ORM) for the Python. It creates SQL database and can create relationships without writing SQL code. It sounds simple but it was not for me.
What is Alembic
Alembic is a database migrations tool that was created by SQLAlchemy's author to help populate the database.
Recap:
SQLAlchemy: Creates a skeleton(structure) of the database.
Alembic: Seeds or populates the data in the database.
First of all, SQLAlchemy was pretty tough to understand but after many trials and errors, I managed to create a basic database using two separate JSON files and two "seed" files to create a skeleton of my database: aircraft.db based on the schema and use a "models" file to populate the actual data using Alembic.
The problems that I ran into were mostly in writing the schema in the seed and models files. SQLAlchemy doesn't allow for incorrect names for database "key" references such as "id" or "user" in the schema between the tables so it gives a whole bunch of error codes.
My "models" for aircraft.db
from sqlalchemy.orm import declarative_base , relationship, backref
from sqlalchemy import Column, Integer, String, ForeignKey
Base = declarative_base()
#Aircraft fleet description
class Aircraft(Base):
__tablename__ = 'Aircraft'
id = Column(Integer, primary_key=True)
make = Column(String)
model = Column(String)
body_type = Column(String)
tasks = relationship("Aircraft_Tasks", backref=backref("Aircraft"))
def __repr__(self):
return f"\n<Aircraft" \
+ f"id={self.id}, " \
+ f"make={self.make}, " \
+ f"model={self.model}, " \
+ f"body_type={self.body_type} " \
+ ">"
#Air Transportation Association (ATA) chapter numbers and tasks
class Aircraft_Tasks(Base):
__tablename__ = 'Maintenance_tasks'
id = Column(Integer, primary_key=True)
ata_chapter_number = Column(Integer)
ata_chapter_name = Column(String)
task = Column(String)
aircraft_id = Column(Integer, ForeignKey('Aircraft.id'))
def __repr__(self):
return f"\n<Aircraft_Tasks " \
+ f"id={self.id}, " \
+ f"ata_chapter_number={self.ata_chapter_number}, " \
+ f"ata_chapter_name={self.ata_chapter_name}, " \
+ f"task={self.task} " \
+ ">"
The "tablename" represents the name of the database table and items below such as the "id" represent the columns as shown above taking a "String" as values. The Columns can take Integers as values as defined on the second table under "ata_chapter_number"
The "repr" is a method that shows a class's object into a string readable form. It makes the class object easy to read and recreate it.
After the database was created, the data was migrated using Alembic.
alembic init "your directory" #initiate data migration
alembic revision --autogenerate -m 'your schema update description' #create a skeleton database file with a descriptor of your version name
alembic upgrade head #upgrade(if different) or populate/migrate the data
After running "alembic upgrade head"
Populated database
After creating and joining the database, I used it in my CLI app, called Aircraft tech taskAid.
This app is a CLI(Command Line Interface) that basically takes user input and manages tasks by adding, deleting tasks from the database as well as adding new tasks to the database.
I have used Simple term Menu package by IngoMeyer441 to have a menu for the initial page. The title was created by using Text to ASCII Art Generator (TAAG) at patorjk.com and prettyCli by Noyoshi to color the different menu titles. It lists options to choose the aircraft, pending work, and quit.
Main menu:
#Initial app menu
def main(self):
while True:
heading.hello_air_tech()
print(green("Welcome Aircraft Tech! Please make a selection:"))
options = ["Aircraft", "Pending work", "Quit"]
menu = TerminalMenu(options)
menu_entry_index = menu.show()
# Aircraft is selected
if menu_entry_index == 0:
self.show_aircraft_models()
# Pending work is selected
elif menu_entry_index == 1:
self.manage_pending_work()
#Check if any pending tasks before exiting the app
else:
if self.pending_tasks:
print(yellow("You have pending tasks. Are you sure you want to quit? (Y/N)"))
choice = input().lower()
if choice == 'y':
print(magenta("Goodbye!"))
break
else:
print(magenta("Goodbye!"))
break
Depending on the option chosen, the app will lead to another submenu such as "Pending work" as shown below.
Displaying all the tasks from database:
def handle_aircraft_tasks(self, selected_model):
while True:
print(blue(f"You have selected aircraft model: {selected_model}"))
tasks = self.session.query(Aircraft_Tasks).all()
if tasks:
task_strings = [f"ATA {task.ata_chapter_number}, {task.task}" for task in tasks]
task_menu = TerminalMenu(task_strings + ["Go Back"], title="Select a Task:")
task_index = task_menu.show()
if task_index >= 0 and task_index < len(tasks):
selected_task = tasks[task_index]
print(red(f"You selected: ATA {selected_task.ata_chapter_number} : {selected_task.task}"))
self.pending_tasks.append(selected_task)
print("Task added to pending work.")
else:
break
else:
print("No tasks available for selected aircraft model.")
break
In this section, after the aircraft is chosen, all the tasks get called by:
tasks = self.session.query(Aircraft_Tasks).all()"
if the aircraft model was chosen by satisfying the while loop condition as True.
This is the result:
After a task is chosen, the selected item is shown:
And also added to the pending work menu:
if task_index >= 0 and task_index < len(tasks):
selected_task = tasks[task_index]
print(red(f"You selected: ATA {selected_task.ata_chapter_number} : {selected_task.task}"))
self.pending_tasks.append(selected_task)
print("Task added to pending work.")
The Pending Work menu has the ability to add, delete, and create new tasks.
def manage_pending_work(self):
while True:
if not self.pending_tasks:
print(red("No pending tasks."))
else:
print("Pending tasks:")
for i, task in enumerate(self.pending_tasks, start=1):
print(f"{i}. ATA {task.ata_chapter_number} : {task.task}")
#Pending tasks submenu
options = ["Add Task", "Remove Task", "Go Back"]
menu = TerminalMenu(options, title="Pending Work:")
option_index = menu.show()
if option_index == 0:
self.add_task_to_pending()
elif option_index == 1:
if not self.pending_tasks:
print("No pending tasks to remove.")
else:
task_strings = [f"ATA {task.ata_chapter_number}, {task.task}" for task in self.pending_tasks]
task_menu = TerminalMenu(task_strings + ["Go Back"], title="Select a Task to Remove:")
task_index = task_menu.show()
if task_index >= 0 and task_index < len(self.pending_tasks):
removed_task = self.pending_tasks.pop(task_index)
print(f"Task removed from pending work: ATA {removed_task.ata_chapter_number} : {removed_task.task}")
self.remove_task_from_database(removed_task) # Remove the task from the database
else:
break
The manage_pending_work function has a condition where it will check if there are any task added the list and will "alert" the user if there are any pending tasks or not. It also has the ability to delete tasks from the database via remove_task_from_database function.
def remove_task_from_database(self, task):
self.session.delete(task)
self.session.commit()
The most important part here is including session and using the delete method and commit function. Otherwise the database won't be changed and will give you a bunch of errors.
There are more functions to the app but these are main ones. Thank you for reading. I hope this was helpful and see you in the next blog.
Top comments (0)