SQLAlchemy Expression Language - Just Enough

Image Credit LOBS Arts

Expression language is a succinct way in SQLAlchemy to interacting with database entities using pythonic construct. Lots can be done through expression language, however, in this post, we'll focus on the Just Enough.

We'll see how to SELECT, INSERT, UPDATE and DELETE from database table using SQLAlchemy expression language. SQLAlchemy's expression language sits within Core layer. Item table defined with CORE API construct from previous guide will be used in this journey.

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.

If you're following the codebase, a little restructuring was done -> Moved DDL operations in db/models/items.py to db/scripts/items_el.py

We'll be working with db/scripts/items_el.py and main.py file in this guide.

SELECT with SQLAlchemy expression language

SELECT-ing from database falls in Database Query Language (DQL) for which we can either select all available columns or selected few from a table.

For this purpose, SQLAlchemy provides a select function which can be imported from sqlalchemy.sql module. If interest is only in a subset of the table column, select function requires that those columns be explicitly specified as parameter on the function call.

To access table's column, sqlalchemy.Table instances has a .c attribute which is a collection of all columns in a table. e.g Item.c returns a collection of all columns in Item table and to access specific column Item.c.column_name can be used.

select_statement = select(Item.c.name, Item.c.id, Item.c.category, ...)

To retrieve all columns, each column can be explicitly specified (like above snippet) or the table instance can be given as is.

select_statement = select(Item)

db/scripts/items_el.py

class DQL:
    """Encapsulates database query language (DML)"""

    @staticmethod
    def retrieve_all_items():
        """Retrieves all data entries and 
        corresponding columns from Item table.
        """
        with create_connection() as conn:
            return conn.execute(select(Item))

Hype
Ignore the class & function construct from above code, the magic is a simple statement select(Item). Others are for structure while conn.execute() is familiar as we've used it previously.

DQL class is created to encapsulate all select query expressions & database calls. retrieve_all_items function return all Item entries and also all columns.

To retrieve a unique Item entry, an upgrade of chaining .where() construct to the existing select construct is all required.

db/scripts/items_el.py

class DQL:
    """Encapsulates database query language (DML)"""

    # codes from existing methods         

    @staticmethod
    def retrieve_item_by_id(id:int):
        """Retrieves a single item by it's id.

        - id <int> A unique identifier of an item.
        """
        statement = (
            select(Item)
            .where(Item.c.id==bindparam('id', type_=Integer))
        )
        with create_connection() as conn:
            result = conn.execute(statement, {'id': id})
            return result

SELECT statement from above snippet utilizes .where() construct which is passed the id (primary key) column of Item table for specificity.

The little catch is bindparam which simply, is used to specify the value we're expecting from user and the type of the value. In this case we're expecting an id of type Integer. This is passed as a dictionary to the second argument of conn.execute.

bindparam is imported from sqlalchemy.sql

INSERT with SQLAlchemy expression language

Database INSERT statement is used to populate a database table. INSERT is a data manipulation language (DML) clause in SQL. Instances of sqlalchemy.Table class has .insert() method to facilitate data insertion to the respective table. On data insertion, we either specify the column name for which we have data & feed the statement those data as values or serve data for all available columns while factoring available constraints.

Statement Construct Point

1. Specify table column for which we've data or address all columns
2. Send data as values to the statement
3. factor constraints

db/scripts/items_el.py

class DML:
    """Encapsulates database manipulation language (DML)"""

    @staticmethod
    def add_item(name:str, category:str):
        """Adds a single item to Item table"""
        statement = Item.insert().values(name=name, category=category)
        with create_connection() as conn:
            conn.execute(statement)
            conn.commit()

Item.insert().values(name=name, category=category) from above code meets all points outlined for the insert statement construct. Item table has 3 columns ( id, name & category ). id column wasn't specified on above code fragment as it's a primary key which autoincrements and by default self populates. We've only supplied name and category column. We could as well omit category column as it is a nullable column and would default to None if nothing is supplied. It's also possible to utilize bindparam and supply the values as dictionary to the second argument to the conn.execute function.

# every other code above

statement = Item.insert().values(
    name=bindparam('name', type_=String),
    category=bindparam('category', type_=String)
)
with create_connection() as conn:
    conn.execute(statement, {'name': name, 'category': category)

# every other code below

With add_item(...) function, only one Item record can be inserted at a time. SQLAlchemy allows for insertion of multiple items which would translates into execute many statement.

db/scripts/items_el.py

class DML:
    """Encapsulates database manipulation language (DML)"""

    # codes from existing methods 

    @staticmethod
    def add_items(payload:List[Dict[str, str]]):
        """Inserts multiple items to Item table

        - payload <list> new data to be added to
                         Item table. Each dict has
                         key mapped to the Item table
                         and it's corresponding value.
        """
        with create_connection() as conn:
            conn.execute(Item.insert(), payload)
            conn.commit()

add_items function takes a sequence of key-value pair as parameter. This parameter is served as second argument to conn.execute() function call. The insert statement didn't have to specify any columns with .values() and as such, it's expected that the key-value pairs within the sequence payload has all the non-nullable field. In this case just name key value pair would be sufficient as id column can self populate and category column is nullable.

DELETE with SQLAlchemy expression language

With DELETE statement, a single, multiple or all records can be removed from the database table. Instances of sqlalchemy.Table class has a .delete() method which can be used to remove existing records from the table. For specificity, .where() construct is called to identify a record.

db/scripts/items_el.py

class DML:
    """Encapsulates database manipulation language (DML)"""

    # code from existing methods

    @staticmethod
    def delete_item(item_id:int):
        """Deletes an item whose id is passed as a 
        parameter

        - item_id <int> Uniquely identifies an item
                        instance
        """
        with create_connection() as conn:
            statement = Item.delete().where(
                Item.c.id==bindparam('id', type_=Integer)
            )
            conn.execute(statement, {'id': item_id})
            conn.commit()

delete_item function takes item_id parameter to identify the record to be deleted from Item table. Item.delete() converts to the statement: delete all records from Item table. Chaining .where() construct helps to specify the given record to be deleted. In this case, the item id column is used for record lookup.

To remove multiple records, although there're couple of ways this could be achieved, we'll utilize .in_() construct.
.in_( ) construct takes a sequence as an argument. This construct is called on column for which records are to be identified by (e.g id column). If any database record matches with given value in the sequence, such records will be selected for deletion (in this case).

db/scripts/items_el.py

class DML:
    """Encapsulates database manipulation language (DML)"""

    # code from existing methods

    @staticmethod
    def delete_many_items_by_id(id:list):
        """Deletes multiple items with the corresponding
        id
        """
        with create_connection() as conn:
            statement = Item.delete().where(Item.c.id.in_(ids))
            conn.execute(statement)
            conn.commit()

delete_many_items_by_id allows for deletion of records whose id are contained in the ids sequence. Below is a breakdown of the statement.

Item.delete()     # delete records in Item table
.where(           # identified by 
   Item.c.id.in_(ids)   # item id found in ids sequence
) 

To delete all records in a database table, pass Item.delete() without .where() construct to conn.execute().

conn.execute(Item.delete())
conn.commit()

Knowing how to perform a complete removal of all entries in a table is crucial but its usecase would be rare going forward in this journey, hence why no function is created for it.

UPDATE with SQLAlchemy expression language

To update existing record, there's need for specificity to avoid the mistake of changing all records with an update meant for just a single record. .where() construct plays a crucial role to avoid such pitfall. UPDATE statement is a data manipulation language (DML).

Instances of sqlalchemy.Table class has .update() method which tells the kind of operation to be performed. To update any value, there's need to be explicit on the column name and the new value by chaining .values() construct.

db/scripts/items_el.py

class DML:
    """Encapsulates database manipulation language (DML)"""

    # code from existing methods

    @staticmethod
    def update_item(item_id:int, data:Dict[str, str]):
        """Updates an existing item

        - item_id <int> Uniquely identifies an item
                        instance

        - data <dict>   Key-value pair with column name
                        as key and the new entry for 
                        column as value.
        """
        with create_connection() as conn:
            statement=Item.update().where(
                Item.c.id==bindparam('item_id', type_=Integer)
            ).values(**data)
            conn.execute(statement, {'item_id': item_id})
            conn.commit()

update_item function takes two parameter, first for identifying record by its id and second parameter which is a key-value pair which would serve as the columns and the new value to make update by. The second parameter is destructured in .values(**data) construct.

Putting It Altogether

Having constructed just enough DML and DQL operations with SQLAlchemy expression language, lets validate the functions and use them within main.py.

Insert Single Record

_main.py_

from db.scripts import queries as q, items_el


if __name__ == "__main__":
    # Add 3 items individually
    # ------------------------

    items_el.DML.add_item(name="Potatoes", category="Meal")
    items_el.DML.add_item(name="Tuna", category="Grocery")
    items_el.DML.add_item(name="PS 5", category="Game")

    # Output available records in the database
    # ----------------------------------------

    results = items_el.DQL.retrieve_all_items()
    for row in results:
        print("{id}) {name:<30s} {category}".format(
                id=row['id'],
                name=row['name'], 
                category=row['category']
            )
        )

Insert Multiple Records Together

_main.py_

from db.scripts import queries as q, items_el


if __name__ == "__main__":
    # Add 4 items together
    # --------------------

    items_el.DML.add_items(
        payload=[
            {'name': 'Addidas ZX 22 Boost', 'category': 'Shoe'},
            {'name': 'Nike Revolution', 'category': 'Shoe'},
            {'name': 'NK Force Dunk', 'category': 'Shoe'},
            {'name': 'Nike Air', 'category': 'Shoe'},
        ]
    )

    # Output available records in the database
    # ----------------------------------------

    results = items_el.DQL.retrieve_all_items()
    for row in results:
        print("{id}) {name:<30s} {category}".format(
                id=row['id'],
                name=row['name'], 
                category=row['category']
            )
        )

Kindly note that the id of every record is outputted, this will be used for DELETE & UPDATE statement.

Delete Single Record

_main.py_

from db.scripts import queries as q, items_el


if __name__ == "__main__":
    # Delete a single item
    # --------------------

    items_el.DML.delete_item(item_id=12) 
    # kindly note that 12 is an id & mightn't correspond to what
    # you have in your output for existing record.


    # Output available records in the database
    # ----------------------------------------

    results = items_el.DQL.retrieve_all_items()
    for row in results:
        print("{id}) {name:<30s} {category}".format(
                id=row['id'],
                name=row['name'], 
                category=row['category']
            )
        )

Delete Multiple Record

_main.py_

from db.scripts import queries as q, items_el


if __name__ == "__main__":
    # Delete a multiple items
    # -----------------------

    items_el.DML.delete_many_items_by_id([19, 20, 21, 22, 23])
    # kindly note that given ids mightn't correspond to what
    # you have in your output for existing record.


    # Output available records in the database
    # ----------------------------------------

    results = items_el.DQL.retrieve_all_items()
    for row in results:
        print("{id}) {name:<30s} {category}".format(
                id=row['id'],
                name=row['name'], 
                category=row['category']
            )
        )

From above code we've specified to delete 5 records with id 19, 20, 21, 22, 23 respectively.

Update a Record

_main.py_

from db.scripts import queries as q, items_el


if __name__ == "__main__":
    # Update single record
    # ---------------------

    items_el.DML.update_item(item_id=12, data={'category': 'Shoe'})

    # Output available records in the database
    # ----------------------------------------

    results = items_el.DQL.retrieve_item_by_id(12)
    for row in results:
        print("{id}) {name:<30s} {category}".format(
                id=row['id'],
                name=row['name'], 
                category=row['category']
            )
        )

Above should output a single record as we've used retrieve_item_by_id() function and specified the id of the record which we're interested in.

Conclusion

SQLALchemy expression language has other interesting constructs such as join, join_from, common table expressions (cte), aggregate functions e.t.c. This guide has uncover the just enough for further explorations.

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

• Codebase on github

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