Database Interaction with SQLAlchemy - Raw DML & DQL

Image credit to Игорь Левченко

In this discovery, we'll look at how to perform SQL DML ( CREATE, UPDATE, DELETE ) and DQL ( SELECT ) operations with SQLAlchemy taking a raw ( writing all SQL query ) approach.

This is a series where I document discoveries in my Journey to SQLAlchemy. If you haven't seen my previous posts, check them out - links are somewhere at the top and bottom of this page.

This journey would solely utilize

• db/scripts/queries.py: contains queries to interact with the database

• main.py: application entrypoint.

To follow along, clone codebase from github

To understand codebase structure, visit SQLAlchemy with Docker - The Structure.

NOTE
This journey is quite technical and requires basic understanding of SQL. This guide focuses on SQLAlchemy processes to interact with databases.

Data Manipulation Language (DML) : with this language, we're able to interact with data in the database through clauses which allows us to add (CREATE), modify (UPDATE) and trash (DELETE) data records.

Data Query Language (DQL) is used to retrieve data from the database. This uses the SELECT clause.

Using SQL SELECT clause in SQLAlchemy

With SELECT clause, we're able to retrieve data from the database. We previously used this clause to retrieve tables available in our database.

Below code fragment contains a function that runs the SELECT clause to retrieve all records from Item table.

db/scripts/queries.py

# ....         other code are above this function

def retrieve_all_item():
    """Retrieves all records from Item table"""
    with create_connection() as conn:
        results = conn.execute(
            text("SELECT * FROM Item")
        )

        for result in results:
            print(result)

When database is queried/interacted with, there're responses containing the effect of the interaction which are returned as an instance of Result object. In this case, we queried the database to return all records in Item table. The returned records are stored on Result instance variable results, automatically created on execution of SELECT statement. From above code, we iterate over results and output each record with print statement.

Using SQL INSERT clause in SQLAlchemy

We currently do not have any record in the database tables, with the help of SQLAlchemy, we'll execute CREATE clause to add some.

db/scripts/queries.py

# ....         other code are above this function

def insert_item(name:str, category:str):
    """Inserts new record into Item table

    param: name [str] Represents an item name to be added
    param: category [str] Represents an item category
    """
    with create_connection() as conn:
        conn.execute(
            text("INSERT INTO Item (name, category) VALUES (:name, :category)"),
            {'name':name, 'category':category}
        )
        conn.commit()

Above code contains a function that adds a single record to Item table. To add a new record, the data would be collected from a source e.g a human user, automation e.t.c. which will be fed through named parameter

Explaining named parameter

conn.execute(
  text("INSERT INTO Item (name, category) VALUES (:name, :category)"),

  # ... collapsed code
)

SQLAlchemy recommends executing an external source (e.g human user, automation process) data using named parameter syntax denoted with :any_given_name when using textual construct. This is important for many reasons, most importantly, it mitigates against SQL injection. The named parameter from above code snippet are :name and :category

How is named parameter value passed ?

conn.execute( 
  text(...),  # statement within text function are collapsed
  {'name':name, 'category':category} )

To populate named parameters, a dictionary is passed as the second argument for conn.execute() function, where the key maps to the name assigned to the named parameters

main.py

from db.scripts.queries import (
    # ........ other imported functions above
    insert_item,
    retrieve_all_item,
)

if __name__ == "__main__":
    insert_item(name="Tesla Model S", category="Auto")
    retrieve_all_item()

We imported insert_item and retrieve_all_item function and utilize them within if block. Running python main.py from terminal to execute the above code would insert some data to Item table and retrieve the inserted data.

SQLAlchemy can also handle multiple insertion with the INSERT clause. To do this, instead of passing a dictionary when feeding data to named parameter, a list of dictionary would be expected, where each dictionary will represent distinct rows and have keys that map appropriately to the expected named parameters. Add the following code to allow for multiple data insertion.

db/scripts/queries.py

from typing import List, Any, Dict

# ....         other code are above this function


def insert_multiple_items(data:List[Dict[str, Any]]):
    """Allows insertion of multiple records into Item table

    param: data [List] sequence of dictionary where each 
           dictionary represents a record to be added to 
           the db
    """
    with create_connection() as conn:
        conn.execute(
            text("INSERT INTO Item (name, category) VALUES (:name, :category)"),
            data
        )
        conn.commit()

To see multiple insertion in practice, we'll utilize insert_multiple_items function within main.py as shown below.

main.py

from db.scripts.queries import (
    # ........ other imported functions above
    insert_multiple_items,
)


if __name__ == "__main__":
    insert_multiple_items([
        {
            'name': 'Iphone 14 Pro Max',   
            'category': 'mobile'
        },
        {
            'name': 'Pizza',
            'category': 'meal'
        }
    ])
    retrieve_all_item()

Running python main.py, would insert two new records into Item table (making total records to be 3 in Item table), thereafter all records would be outputted. Your result should be similar to that of the image below.

Using SQL UPDATE clause in SQLAlchemy

With an UPDATE clause, existing record can be modified. To update an existing record, the unique identifier for said record is required for explicit identification of the record.

We currently have three (3) records in Item table, we'll update the name value of the first record (with a unique identifier 1) from Tesla Model S to Tesla Starlink.

Below code shows function that processes an update on Item record existing in the database.

db/scripts/queries.py

# ....         other code are above this function

def update_existing_item_name(item_id:int, new_value:Any):
    """Handles the update of an existing record name
    in `Item` table

    param: item_id [int] A unique identifier of the record to be 
           updated.
    param: new_value [Any] Replacement value for previous `name` column value.
    """
    with create_connection() as conn:
        conn.execute(
            text("UPDATE Item SET name=:update_value WHERE id=:id"),
            {'id': item_id,  'update_value': new_value}
        )
        conn.commit()

Utilize this function within main.py by first importing it and then call it in the codebase as show below.

main.py

from db.scripts.queries import (
    # ... there're other import statement above
    update_existing_item_name,
)

if __name__ == "__main__":
    update_existing_item_name(item_id=1, new_value="Tesla Starlink")
    retrieve_all_item()

Running python main.py would update the previous value from "Tesla Model S" -> to Tesla Starlink for record with of
a unique identifier of 1. Note that an error would be raised for an unexistent identifier.

Using SQL DELETE clause in SQLAlchemy

DELETE clause handles removal of existing record from the database. Just like UPDATE clause, to delete a unique record, DELETE clause would require the unique identifier of record to be deleted. Below code shows an example function utilizing DELETE clause.

db/scripts/queries.py

def delete_item(id:int):
    """Removes an existing item record from `Item` table.

    param: id [int] A unique identifier for item to be deleted
    """
    with create_connection() as conn:
        conn.execute(text("DELETE FROM Item WHERE id=:id"), {'id':id})
        conn.commit()

When delete_item function is called within main.py, it deletes record whose id is given as a unique identifier on the function call.

main.py

from db.scripts.queries import (
    # ... there're other import statement above
    delete_item,
)

if __name__ == "__main__":
    delete_item(1)
    retrieve_all_item()

The above should delete an item whose unique identifier is 1 ( 1 is passed to delete_item() function call). If an unexistent identifier is given, an error would be raised.

Conclusion

SQLAlchemy at it's raw state (using text function and writing all queries) provides a very easy to follow process/structure to interacting with the database. There're little to no changes with construct when using DML, DQL, DDL clauses while interacting with the database.

Thank you for being AWESOME

✍️

• Encountered any bugs, typos or concept misinterpretation from this guide ? Let me know in the comment section below.

❤️

• Hit the LIKE button and FOLLOW me so you're notified on future discoveries.

☕

• A lot is accomplished with a cup of coffee

Thank you for your time and for staying with me on this one.