Image Credit to winterseitler
Code available on Github
In this discovery, we'll use SQLAlchemy to perform fundamental database operations. The database operations performed in this guide will utilize raw SQL query. We'll manually write and run all database operations using SQLAlchemy.
This is a series where I document discoveries in my Journey to SQLAlchemy. If you haven't seen my previous posts, check them out - they are somewhere at the top and bottom of this page.
NOTE
This journey is quite technical. Basics understanding of SQL is a requirement. Only shallow touches will be given to SQL query syntax on this guide as our focus is on SQLAlchemy.
Database as a data store supports different languages to construct and interact with the stored data. Below is a brief intro to the languages:
Data Query Language (DQL): This language helps to retrieve stored data e.g. this blog post is stored on a database. If you're reading this post, you've triggered a DQL to the database. SQL SELECT statement is used to query data from the database.
Data Definition Language (DDL): This language constructs the structure to manage data. Without Data Definition Language, there wouldn't be a library to save or retrieve data from. SQL statements for data definition in a database are CREATE, DROP, RENAME, TRUNCATE, ALTER.
Data Manipulation Language (DML): While DDL interacts with structure, DML interacts with the actual data. SQL statement for this language allows for data INSERT, UPDATE, or DELETE query.
Codebase Structure
To setup codebase structure, visit SQLAlchemy with Docker - The Structure
Maintaining codebase structure, we'll create a new folder within db folder and we'll name it scripts. Add two files within db/scripts/, __init__.py and queries.py respectively.
We'll be working with two files in this journey
db/scripts/queries.py: This file will contain all of our database queries with which we'll be interacting with the database.main.py: We'll use this file to run the program containing database queries.
Case Study
Lets assume, each time we have a craving, we get to track it and update any of the tracked record after the crave has been satisfied.
Table blueprint
ITEM
| id | name | timestamp | category |
|---|---|---|---|
| 1 | Tesla Model S | 07-12-2022 00:00:00 | auto |
| 2 | Iphone 14 Pro Max | 03-29-2022 00:00:00 | mobile |
| 3 | Pizza | 01-12-2022 00:00:00 | meal |
CRAVEDITEM
| id | item_id | date_tracked | is_satisfied |
|---|---|---|---|
| 1 | 1 | 10-12-2022 00:00:00 | False |
| 2 | 2 | 09-29-2022 00:00:00 | True |
| 3 | 2 | 10-12-2022 00:00:00 | False |
There are two tables in the blueprint - Item and CravedItem, each having 4 columns.
Lets use Data Definition Language (DDL) to define these structure where we'll track our records. This structure is called table in database world.
CREATE statement in action
In db/scripts/queries.py add the following code
from sqlalchemy import text
from db.core.initializer import create_connection
def create_item_table():
"""Creates Item table.
col: id Unique identifier - primary key.
col: timestamp datetime defaults to current
date and time the item is tracked.
col: name Name of item - field is required.
col: category Category of the tracked item
"""
# Creates a connection to the database
with create_connection() as conn:
# Utilizes connection instance to execute SQL query
conn.execute(
text(
'''
CREATE TABLE Item (
id SERIAL PRIMARY KEY,
date_tracked TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
name VARCHAR(250) NOT NULL,
category VARCHAR(200),
)
'''
)
)
# Sends the query to the database
conn.commit()
The first two lines imports, text from sqlalchemy and create_connection from database initializer that we created at journey's setup.
We define a create_item_table function. This function would create Item table with the exact columns as outlined on the blueprint with the help of connection instance returned by create_connection.
conn.execute( .... )
conn.execute( ... ) is non-autocommitting which means the SQL query taken by which is whatever is in text(...) is not automatically processed. We've to manually process each query execution using conn.commit().
text is a function which takes raw sql query. text function call returns an executable object which will be executed against the database by conn.execute() only when conn.commit is called
In main.py import create_item_table from db/scripts/queries.py with below code
from db.scripts.queries import create_item_table
Replace run_db_select_statement() (from previous guide) with create_item_table(). main.py file should now look like this
# .... import statements
if __name__ == "__main__":
create_item_table()
Run python main.py from your terminal, you should have an output similar to what the image below holds
As an experiment, re-run python main.py, this should give you an error as there cannot be two Tables with similar name. Outputted error should be similar to the below.
DROP statement in action
DROP is a critical database clause/statement and should be used with caution. When used, it'll teardown/destroy the entity and whatever is in such entity. We created an Item table structure earlier, executing a DROP statement on that table will not only destroy the structure and make it unavailable, it'll also destroy the records in it.
Open up db/scripts/queries.py file and include the code below which is a new function that executes DROP statement on Item table.
# .... other code are above this function
def drop_item_table():
"""Delete Item table"""
with create_connection() as conn:
conn.execute(text("DROP TABLE Item"))
conn.commit()
Open up main.py file and import drop_item_table function. Your import statement should now look like
from db.scripts.queries import create_item_table, drop_item_table
Replace create_item_table function call to drop_item_table function like below
# .... other code are above this function
if __name__ == "__main__":
# create_item_table()
drop_item_table()
Open your terminal and run python main.py, this should delete the existing Item table. If you should uncomment create_item_table and comment out drop_item_table, making your code like so
# .... other code are above this function
if __name__ == "__main__":
create_item_table()
# drop_item_table()
Running python main.py now wouldn't give you the duplicate table error we had earlier as the existing table was dropped, instead you should now have the Item table recreated.
ALTER X RENAME statement in action
There will be less fun using RENAME statement if we can't see existing table. Open db/scripts/queries.py and include below code to retrieve (DQL) all available tables that are not autogenerated on database creation.
# .... other code are above this function
def show_all_tables():
"""Show all available tables in the database
Returned tables excludes those having their
schema as `pg_catalog` and `information_schema`.
"""
with create_connection() as conn:
results = conn.execute(
text(
'''
SELECT * FROM pg_catalog.pg_tables
WHERE schemaname != 'pg_catalog'
AND schemaname != 'information_schema'
'''
)
)
for data in results:
# Only print the table name contained on Index 1
print(f"{data[1]} Table".title())
Import this function into main.py, and call it within the if block like below
from db.scripts.queries import (
create_item_table,
drop_item_table,
show_all_tables
)
if __name__ == "__main__":
# create_item_table()
# drop_item_table()
show_all_tables()
Running python main.py should emit the existing tables in the database
RENAME statement is used to change the name of an existing Table e.g if we want to change Item table name to Commodity, we'll use the RENAME statement.
Open up db/scripts/queries.py and include the following function which will rename existing Item table to Commodity
# .... other code are above this function
def rename_item_table():
"""Rename Item table if it exists in the database"""
with create_connection() as conn:
conn.execute(
text(
'''
ALTER TABLE Item
RENAME TO Commodity
'''
)
)
conn.commit()
Now that we have rename_item_table function, import it into main.py and update if block
from db.scripts.queries import (
create_item_table,
drop_item_table,
show_all_tables,
rename_item_table
)
if __name__ == "__main__":
show_all_tables()
rename_item_table()
show_all_tables()
The if block code statement will display the available tables before and after execution of renaming Item table. After running above code using python main.py, we would no longer have a table named Item as it would've been renamed to Commodity.
Run above command using python main.py and you should have similar result
Exercise
Rename the table fromCommoditytoItem. Do note that we'll be working withItemtable for future discoveries, ensure to complete this exercise
NOTE
SQLALchemy doesn't have a general support for ALTER & some other DDL statement
Don't forget me !!!
From our table blueprint, we had two tables, but so far we have only created one.
Open up db/scripts/queries.py and include this function. First function holds query to create CravedItem table and the second function holds query to drop CravedItem table
# .... other code are above this function
def create_craved_item_table():
"""Create CravedItem table.
This table keeps tracks of items that are craved for
col: id Unique identifier - primary key.
col: item_id Item unique identifier - field is
required.
col: date_tracked datetime defualts to current
date and time the item is tracked.
col: is_satisfied Denotes either or not craved item
has been satisfied.
"""
# Creates a connection to the database
with create_connection() as conn:
# Utilizes connection instance to execute SQL query
conn.execute(
text(
'''
CREATE TABLE CravedItem (
id SERIAL PRIMARY KEY,
item_id INT NOT NULL,
date_tracked TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
is_satisfied BOOLEAN DEFAULT False,
CONSTRAINT fk_item FOREIGN KEY(item_id) REFERENCES Item(id)
)
'''
)
)
# Sends the query to the database
conn.commit()
def drop_craved_item_table():
"""Delete CravedItem table"""
with create_connection() as conn:
conn.execute(text("DROP TABLE CravedItem"))
conn.commit()
CravedItem table relates to Item table through FOREIGN KEY constraint, because of this, drop_item_table DDL query will fail at execution. This would warrant an update in the DROP statement used in destroying Item table
conn.execute(text("DROP TABLE Item CASCADE"))
I only just showed the part that has been updated. The only difference is the CASCADE that's added
In summary, we've seen how to use CREATE, DROP, ALTER X RENAME DDL Clause to perform database interaction through SQLAlchemy. In addition we've maintained structured codebase by following the principle of Separation of Concern.
SQLAlchemy has more to offer
Although this raw query approach we've just taken has it's place of creating that awareness that this is possible in SQLAlchemy, by no means is this how SQLAlchemy is used in the real world on a day to day basis.
Thanks for sticking with me on this discovery. If you like this, don't forget to Bookmark it for later. Hit the Heart ❤️ button and do well to FOLLOW me so you're notified on the next discovery in this journey.
Code available on Github
Top comments (0)