Database Interaction with SQLAlchemy - Raw DDL Tour

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

CRAVEDITEM

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 from Commodity to Item. Do note that we'll be working with Item table 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.

☕ My Coffee Link

Code available on Github