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.pytodb/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 statementselect(Item). Others are for structure whileconn.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
- Specify table column for which we've data or address all columns
- Send data as values to the statement
- 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
idof 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.
☕
Thank you for your time and for staying with me on this one.
Top comments (2)
Did you test any of this post? I found a bunch of typos, etc. delete_many_items_by_id needs to have the same variable (ids) in the signature and the Item.delete statement. items_el.py needs to import create_connection from db.core.initializer, select from sqlalchemy.sql, and Integer from sqlalchemy. Otherwise I have enjoyed your posts; this last one provided an interesting exercise in debugging...
Thanks @guarin1949 for taking the time to read through my posts. I also do appreciate your feedback 🙂. I did run through the reported concerns and yes, there was an issue with the argument name which has been updated now. While you consider the debugging as an interesting one, sorry you had to deal with that mismatch.
This article is a series in which your concerns about the import statement has been treated in previous post -- here are the references:
ref to
create_connectionimportAbove quote is an extract from this article ( On this one, I'll be more explicit in my future articles as the provided block code snippet didn't capture it but i made a statement on it and also the git repo reflected it)
Thanks again Warren for the comment and report. You just made this article better by reporting the observed issues. 🙏
here is a link to github for the complete source code
github.com/spaceofmiah/practical-r...