CRUD operations in SQLAlchemy ORM

Creating a session

When using the SQLAlchemy ORM, interaction with the database takes place through a Session object. It also captures the database connection and transactions. A transaction implicitly starts as soon as Session starts communicating with the database and stays open until Session is committed, rolled back, or closed. You can use the Session class from sqlalchemy.orm to create a session object.

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
engine = create_engine("postgresql+psycopg2://postgres:[email protected]/sqlalchemy_tuts")
session = Session(bind=engine)

You will need to create a Session object every time you interact with the database. The Session constructor takes a certain number of arguments that determine how it works. If you create a session this way, the Session constructor will need to be called with the same set of parameters. To simplify this process, SQLAlchemy provides a sessionmaker class that creates a Session class with default constructor arguments.

from sqlalchemy.orm import Session, sessionmaker
session = sessionmaker(bind=engine)

You just need to call sessionmaker once in the global scope. Once you access this Session class once, you can create instances of it any number of times without passing parameters.

session = Session()

Note that the Session object doesn’t immediately establish a connection to the database. It only happens with the first request.

Inserting (adding) data

To create a new record with SQLAlchemy ORM you need to perform the following steps:

1. Create object
2. Add it to a session
3. Save session

Create two new Customer objects:

c1 = Customer(
first_name = 'Dmitriy',
last_name = 'Yatsenko',
username = 'Moseend',
email = '[email protected]'
)
c2 = Customer(
first_name = 'Valeriy',
last_name = 'Golyshkin',
username = 'Fortioneaks',
email = '[email protected]'
)
print(c1.first_name, c2.last_name)
session.add(c1)
session.add(c2)
print(session.new)
session.commit()

First output: Dmitriy Golyshkin. Two objects have been created. You can access their attributes with the point operator (.). Then the objects will be added to the session.

session.add(c1)
session.add(c2)

But adding objects does not affect the recording in the database, but only prepares the objects for saving in the next commit. You can check this by getting the primary keys of the objects. The id attribute of both objects is None. This means that they are not yet saved in the database. Instead of adding one object at a time, you can use the add_all() method. It takes a list of objects that will be added to the session.

session.add_all([c1, c2])

Adding an object to a session multiple times does not cause errors. You can look at existing objects at any time with session.new.

IdentitySet([, ])

Finally, the commit() method is used to save the data:

session.commit()

After the transaction is saved, the connection resources referenced by the Session object are returned to the connection pool. Subsequent operations will be executed in a new transaction. Now the Customer table looks like this: So far, customers have not purchased anything. So c1.orders and c2.orders will return an empty list.

[] []

Let’s add more customers to the customers table:

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
engine = create_engine("postgresql+psycopg2://postgres:[email protected]/sqlalchemy_tuts")
session = Session(bind=engine)
c3 = customer(
first_name = "Vadim",
last_name = "Moiseenko",
username = "Antence73",
email = "[email protected]"
)
c4 = Customer(
first_name = "Vladimir",
last_name = "Belousov",
username = "Andescols",
email = "[email protected]"
)
c5 = Customer(
first_name = "Tatyana",
last_name = "Khakimova",
username = "Caltin1962",
email = "[email protected]"
)
c6 = Customer(
first_name = "Pavel",
last_name = "Arnautov",
username = "Lablen",
email = "[email protected]"
)
session.add_all([c3, c4, c5, c6])
session.commit()

Also add products to the table items:

i1 = Item(name = 'Chair', cost_price = 9.21, selling_price = 10.81, quantity = 5)
i2 = Item(name = 'Pen', cost_price = 3.45, selling_price = 4.51, quantity = 3)
i3 = Item(name = 'Headphone', cost_price = 15.52, selling_price = 16.81, quantity = 50)
i4 = Item(name = 'Travel Bag', cost_price = 20.1, selling_price = 24.21, quantity = 50)
i5 = Item(name = 'Keyboard', cost_price = 20.1, selling_price = 22.11, quantity = 50)
i6 = Item(name = 'Monitor', cost_price = 200.14, selling_price = 212.89, quantity = 50)
i7 = Item(name = 'Watch', cost_price = 100.58, selling_price = 104.41, quantity = 50)
i8 = Item(name = 'Water Bottle', cost_price = 20.89, selling_price = 25, quantity = 50)
session.add_all([i1, i2, i3, i4, i5, i6, i7, i8])
session.commit()

Create orders:

o1 = Order(customer = c1)
o2 = Order(customer = c1)
line_item1 = OrderLine(order = o1, item = i1, quantity = 3)
line_item2 = OrderLine(order = o1, item = i2, quantity = 2)
line_item3 = OrderLine(order = o2, item = i1, quantity = 1)
line_item3 = OrderLine(order = o2, item = i2, quantity = 4)
session.add_all([o1, o2])
session.new
session.commit()

In this case, only Order objects(o1 and o2) are added to the session. Order and OrderLine are connected by a one-to-many relation. Adding an Order object to the session implicitly adds OrderLine objects as well. But even if you add the latter manually, there will be no error. Instead of passing an Order object when creating an OrderLine instance, you can do the following:

o3 = Order(customer = c1)
orderline1 = OrderLine(item = i1, quantity = 5)
orderline2 = OrderLine(item = i2, quantity = 10)
o3.line_items.append(orderline1)
o3.line_items.append(orderline2)
session.add_all([o3,])
session.new
session.commit()

After the commit, the orders and order_lines tables will look like this: If you now access the orders attribute of the Customer object, a non-empty list will be returned.

[, ]

On the other side of the relationship, you can access the Customer object that the order belongs to via the customer attribute of the Order object – o1.customer. Right now, customer c1 has three orders. To see all the items in the order you need to use the line_items attribute of the Order object.

c1.orders[0].line_items, c1.orders[1].line_items
([, ], [, ])

To get the item of the order use item.

for ol in c1.orders[0].line_items:
ol.id, ol.item, ol.quantity
print('-------')
for ol in c1.orders[1].line_items:
ol.id, ol.item, ol.quantity

Conclusion:

(1, , 3)
(2, , 2)
-------
(3, , 1)
(4, , 4)

All this is possible thanks to relationship() models.

Retrieving data

The query() method of the session object is used to query the database. It returns an object of type sqlalchemy.orm.query.Query, which is simply called Query. It represents the SELECT instruction that will be used to query the database. The following table lists common methods of the Query class. MethodDescriptional()Returns the query result ( Query object) as a listcount()Returns the total number of records in the queryfirst()Returns the first result from the query or None if no recordscalar()Returns the first column of the first record or None if the result is empty. If there are multiple records, throws a MultipleResultsFoundoneexception Returnsa single record. If there is more than one, throws a MutlipleResultsFound exception. If no data, throws NoResultFoundget(pk)Returns the object by primary key(pk) or None if no object was foundfilter(*criterion)Returns a Query instance after applying the WHERElimit(limit)Returns a Query instance after applying the LIMIToffset(offset)Returns a Query instance after applying the OFFSETorder_by(*criterion)Returns a Query instance after applying the ORDER BY statementjoin(*props, **kwargs)Returns a Query instance after creating an SQL INNER JOINouterjoin(*props, **kwargs)Returns a Query instance after creating an SQL LEFT OUTER JOINgroup_by(*criterion)Returns a Query instance after adding a GROUP BY statement to a query

All() method

In its basic form, the query() method takes one or more model classes or columns as arguments. The following code will return all records from the customers table.

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
engine = create_engine("postgresql+psycopg2://postgres:[email protected]/sqlalchemy_tuts")
session = Session(bind=engine)
print(session.query(Customer).all())

[,
,
,
,
,
]

You can also retrieve records from items and orders tables. To get the raw SQL used to run a query on the database, apply sqlalchemy.orm.query.Query as follows: print(session.query(Customer)).

SELECT
customers. ID AS customers_id,
customers.first_name AS customers_first_name,
customers.last_name AS customers_last_name,
customers.username AS customers_username,
customers.email AS customers_email,
customers.address AS customers_address,
customers.town AS customers_town,
customers.created_on AS customers_created_on,
customers.updated_on AS customers_updated_on
FROM
customers

Calling the all() method on a large result object is not very efficient. Instead, you should use the for loop to loop through the Query object:

q = session.query(Customer)
for c in q:
print(c.id, c.first_name)

Previous queries have returned data from all columns in the table. You can prevent this by passing column names explicitly to the query() method:

print(session.query(Customer.id, Customer.first_name).all())

Output:

[(1, 'Dmitriy'),
(2, 'Valeriy'),
(3, 'Vadim'),
(4, 'Vladimir'),
(5, 'Tatyana'),
(6, 'Pavel')]

Note that each list item is a tuple, not a model instance.

The count() method

count() returns the number of elements in the result.

session.query(Item).count()
# output - 8

The first() method

first() method returns the first result of the request or None if the last one returned no data.

session.query(Order).first()
# output - Order:1

The get() method

get() method returns an instance with the corresponding primary key, or None if no such object was found.

session.query(Customer).get(1)
# Output - Customer:1-Moseend

Filter() method

This method allows you to filter results by adding the WHERE operator. It accepts the column, operator, and value. For example:

session.query(Customer).filter(Customer.first_name == 'Vadim').all()

This query will return all customers whose name is Vadim. And here is the SQL equivalent of this query:

print(session.query(Customer).filter(Customer.first_name == 'Vadim'))

SELECT
customers.id AS customers_id,
customers.first_name AS customers_first_name,
customers.last_name AS customers_last_name,
customers.username AS customers_username,
customers.email AS customers_email,
customers.address AS customers_address,
customers.town AS customers_town,
customers.created_on AS customers_created_on,
customers.updated_on AS customers_updated_on
FROM
customers
WHERE
customers.first_name = %(first_name_1)s

The string %(first_name_1)s in the WHERE statement is the placeholder that will be replaced with the actual value(Vadim) when the query is executed. You can pass several filters to the filter() method and they will be combined using the AND operator. For example:

session.query(Customer).filter(Customer.id <= 5, Customer.last_name == "Arnautov").all()

This query will return all customers whose primary key is less than or equal to 5 and whose last name starts with “Ar”.

session.query(Customer).filter(Customer.id <= 5, Customer.last_name.like("Ar%")).all()

Another way to combine conditions is with conjunctions(and_(), or_(), and not_()). Some examples:

# all clients named Vadim and Tatyana
session.query(Customer).filter(or_(
Customer.first_name == 'Vadim',
Customer.first_name == 'Tatyana'
)).all()
# find all with first and last names NOT Yatsenko
session.query(Customer).filter(and_(
Customer.first_name == 'Pavel',
not_(
Customer.last_name == 'Yatsenko',
)
)).all()

The following list demonstrates how to use common comparison operators with the filter() method.

IS NULL

session.query(Order).filter(Order.date_placed == None).all()

IS NOT NULL

session.query(Order).filter(Order.date_placed != None).all()

IN

session.query(Customer).filter(Customer.first_name.in_(['Pavel', 'Vadim'])).all()

NOT INT

session.query(Customer).filter(Customer.first_name.notin_(['Pavel', 'Vadim'])).all()

BETWEEN

session.query(Item).filter(Item.cost_price.between(10, 50)).all()

NOT BETWEEN

session.query(Item).filter(not_(Item.cost_price.between(10, 50))).all()

LIKE

session.query(Item).filter(Item.name.like("%r")).all()

The like() method does a case-sensitive search. Use ilike() to search for matches that are not case-sensitive.

session.query(Item).filter(Item.name.ilike("w%")).all()

NOT LIKE

session.query(Item).filter(not_(Item.name.like("W%")).all()

Limit() method

The limit() method adds a LIMIT operator to the query. It takes the number of records to return.

session.query(Customer).limit(2).all()
session.query(Customer).filter(Customer.username.ilike("%Andes")).limit(2).all()

SQL equivalent:

SELECT
customers. id AS customers_id,
customers.first_name AS customers_first_name,
customers.last_name AS customers_last_name,
customers.username AS customers_username,
customers.email AS customers_email,
customers.address AS customers_address,
customers.town AS customers_town,
customers.created_on AS customers_created_on,
customers.updated_on AS customers_updated_on
FROM
customers
LIMIT %(param_1)s 

Offset() method

The offset() method adds the OFFSET operator to the query. It takes the offset value as an argument. It is often used with limit() operator.

session.query(Customer).limit(2).offset(2).all()

SQL equivalent:

SELECT
customers. ID AS customers_id,
customers.first_name AS customers_first_name,
customers.last_name AS customers_last_name,
customers.username AS customers_username,
customers.email AS customers_email,
customers.address AS customers_addrees,
customers.town AS customers_town,
customers.created_on AS customers_created_on,
customers.updated_on AS customers_updated_on
FROM
customers
LIMIT %(param_1)s OFFSET %(param_2)s

Order_by() method

The order_by() method is used to sort the result using the ORDER BY operator. It takes the names of the columns by which you want to sort the result. By default, it sorts in ascending order.

session.query(Item).filter(Item.name.ilike("wa%")).all()
session.query(Item).filter(Item.name.ilike("wa%")).order_by(Item.cost_price).all()

To sort in descending order, use desc():

from sqlalchemy import desc
session.query(Item).filter(Item.name.ilike("wa%")).order_by(desc(Item.cost_price)).all()

The join() method

The join() method is used to create an SQL INNER JOIN. It takes the name of the table with which you want to execute the SQL JOIN. Use join() to find all customers that have at least one order.

session.query(Customer).join(Order).all()

SQL equivalent:

SELECT
customers.id AS customers_id,
customers.first_name AS customers_first_name,
customers.last_name AS customers_last_name,
customers.username AS customers_username,
customers.email AS customers_email,
customers.address AS customers_address,
customers.town AS customers_town,
customers.created_on AS customers_created_on,
customers.updated_on AS customers_updated_on
FROM
customers
JOIN orders ON customers.id = orders.customer_id

This operator is often used to retrieve data from one or more tables in a single query. For example:

session.query(Customer.id, Customer.username, Order.id).join(Order).all()

You can create an SQL JOIN for more than two tables by combining multiple join() methods as follows:

session.query(Table1).join(Table2).join(Table3).join(Table4).all()

Here is another example that uses 3 joins to find all items in the first order Dmitriy Yatsenko.

session.query(
Customer.first_name,
Item.name,
Item.selling_price,
OrderLine.quantity
).join(Order).join(OrderLine).join(Item).filter(
Customer.first_name == 'Dmitriy,
Customer.last_name == 'Yatsenko',
Order.id == 1,
).all()

The outerjoin() method

Method outerjoin() works like join() but creates LEFT OUTER JOIN.

session.query(
Customer.first_name,
Order.id,
).outerjoin(Order).all()

In this query, the left table is customers. This means it will return all records from customers and only those that match the condition from orders. You can create a FULL OUTER JOIN by passing full=True in the method. For example:

session.query(
Customer.first_name,
Order.id,
).outerjoin(Order, full=True).all()

Group_by() method

Results are grouped using group_by(). This method takes one or more columns and groups the records according to the values in the column. The following query uses join() and group_by() to calculate the number of orders made by Dmitriy Yatsenko.

from sqlalchemy import func
session.query(func.count(Customer.id)).join(Order).filter(
Customer.first_name == 'Dmitriy',
Customer.last_name == 'Yatsenko',
).group_by(Customer.id).scalar()

The having() method To filter results based on the values returned by aggregation functions, the having() method is used, which adds the HAVING operator to the SELECT instruction. Similar to where(), it takes a condition.

session.query(
func.count("*").label('username_count'),
Customer.town
).group_by(Customer.username).having(func.count("*") > 2).all()

Working with duplicates To work with duplicate records you can use the DISTINCT parameter. It can be added to SELECT using the distinct() method. For example:

from sqlalchemy import distinct
session.query(Customer.first_name).filter(Customer.id < 10).all()
session.query(Customer.first_name).filter(Customer.id < 10).distinct().all()
session.query(
func.count(distinct(Customer.first_name))
func.count(Customer.first_name)
).all()

DISTINCT The cast ( ) function of the sqlalchemy library is a common operation that converts data from one type to another.

from sqlalchemy import cast, Date, distinct, union
session.query(
cast(func.pi(), Integer),
cast(func.pi(), Numeric(10,2))
cast("2010-12-01", DateTime),
cast("2010-12-01", Date),
).all()

Merges The union() method of a Query object is used to combine queries. It accepts one or more queries. For example:

s1 = session.query(Item.id, Item.name).filter(Item.name.like("Wa%"))
s2 = session.query(Item.id, Item.name).filter(Item.name.like("%e%"))
s1.union(s2).all()

[(2, 'Pen'),
(4, 'Travel Bag'),
(3, 'Headphone'),
(5, 'Keyboard'),
(7, 'Watch'),
(8, 'Water Bottle')]

By default, union() removes all repetitive entries from the result. Use union_all() to save them.

s1.union_all(s2).all()

Updating data

To update an object, simply set a new value to the attribute, add the object to the session, and save it.

i = session.query(Item).get(8)
i.selling_price = 25.91
session.add(i)
session.commit()

Only one object at a time can be updated this way. To update multiple records at a time, use the update() method of the Query object. It returns the total number of updated records. For example:

session.query(Item).filter(
Item.name.ilike("W%")
).update({"quantity": 60}, synchronize_session='fetch')
session.commit()

Deleting Data

To delete an object, use the delete() method of the session object. It accepts the object and marks it as deleted for the next commit.

i = session.query(Item).filter(Item.name == 'Monitor').one()
session.delete(i)
session.commit()

This commit removes Monitor from the items table. To delete multiple records at once, use the delete() method of the Query object.

session.query(Item).filter(
Item.name.ilike("W%")
).delete(synchronize_session='fetch')
session.commit()

This commit removes all items whose name begins with W.

Raw queries

The ORM allows you to use raw SQL queries with the text() function. For example:

from sqlalchemy import text
session.query(Customer).filter(text("first_name = 'Vladimir'")).all()
session.query(Customer).filter(text("username like 'Cal%'").all()
session.query(Customer).filter(text("username like 'Cal%'")).order_by(text("first_name, id desc")).all()

Transactions

A transaction is a way of executing a set of SQL instructions so that either all or none of them are executed together. If at least one instruction in a transaction fails, the database reverts to its previous state. There are two orders in the database, there are steps in the order shipping process:

1. The date_placed column of the orders table sets the shipment date.
2. The number of items ordered is subtracted from the items.

Both steps must be performed as one to ensure that the data in the tables is correct. In the following code, define a dispatch_order() method that takes order_id as an argument and performs the tasks described above in a single transaction.

def dispatch_order(order_id):
# check if order_id is specified correctly
order = session.query(Order).get(order_id)
if not order:
raise ValueError("Invalid order_id: {}.".format(order_id))
try:
for i in order.line_items:
i.item.quantity = i.item.quantity - i.quantity
order.date_placed = datetime.now()
session.commit()
print("Transaction complete.")
except IntegrityError as e:
print(e)
print("rollback...")
session.rollback()
print("Transaction failed.")
dispatch_order(1)

The first order contains 3 chairs and 2 pens. the dispatch_order() with order_id 1 will produce the following output:

The transaction is now complete.