How Do I …¶
In this section you can find how to do some of the most common database tasks
using Alchemical and SQLAlchemy. All examples assume that an Alchemical
instance has been created and stored in a global variable named db
.
… create a database model?¶
The Alchemical
package provides a Model
class to be used as a base
class for database models:
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from alchemical import Model
class User(Model):
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(128))
The base Model
class is also accessible as a db.Model
attribute of the
Alchemical
instance.
To learn more about how to define database models, consult the SQLAlchemy ORM documentation.
… create a database session?¶
The recommended way to work with a database session is to use a context manager:
with db.Session() as session:
# work with the session here
If you are using the asynchronous version of Alchemical, create a session as follows:
async with db.Session() as session:
# work with the session here
When the session is created in this way, a database transaction is
automatically initiated when required, and the session.flush()
,
session.commit()
and session.rollback()
methods can be used inside the
context manager block as needed. The session is automatically closed and
returned to the session pool when the context manager block ends.
You can also create a session without the context manager and close it manually:
session = db.Session()
# work with the session here
session.close()
And for the asynchronous version:
session = db.Session()
# work with the session here
await session.close()
… start a database transaction?¶
See the previous question for implicitly starting a transaction when one is
needed. To create a session and explicitly start a transaction on it at the
same time, use the db.begin()
method:
with db.begin() as session:
# work with the session here
Or for the asynchronous version of Alchemical:
async with db.begin() as session:
# work with the session here
The transaction is automatically committed if the context manager block completes successfully, or rolled back if an error occurs. The session is then closed.
In cases where a session has already been created, the begin()
method can
be called on it to explicitly start a transaction:
with session.begin():
# work with the session here
Or with the asynchronous version:
async with session.begin():
# work with the session here
As in the previous example, the transaction is committed on success, or rolled back on error. The session in this case is not closed.
Calling begin()
on a session object only works if an implicit transaction
hasn’t been started on the session yet. For a session that already has an
active transaction, the begin_nested()
method can be used to create a
nested transaction or save point:
with session.begin_nested():
# work with the session here
Or with the asynchronous version:
async with session.begin_nested():
# work with the session here
… save an object to a database table?¶
To add a new object to the database, use session.add()
:
# with a transaction's implicit commit
with db.begin() as session:
new_user = User(name='mary')
session.add(new_user)
# with an explicit commit
with db.Session() as session:
new_user - User(name='mary')
session.add(new_user)
session.commit()
If you are using the asynchronous version of Alchemical:
# with a transaction's implicit commit
async with db.begin() as session:
new_user = User(name='mary')
session.add(new_user)
# with an explicit commit
async with db.Session() as session:
new_user = User(name='mary')
session.add(new_user)
await session.commit()
… retrieve an object by its primary key?¶
The session.get()
method can be used to retrieve an object by its primary
key:
# retrieve user with id=2
with db.Session() as session:
user = session.get(User, 2)
With the asynchronous version:
# retrieve user with id=2
async with db.Session() as session:
user = await session.get(User, 2)
… execute a database query?¶
Use the session.execute()
, session.scalars()
or session.scalar()
methods, depending on your needs. Example:
# find all users with names starting with "m"
with db.Session() as session:
for user in session.scalars(User.select().where(User.name.like('m%'))):
print(user.name)
With the asynchronous version the session.stream()
and
session.stream_scalars()
methods can also be used, with the difference that
unlike the above methods, these do not buffer all results in memory, so they
are more efficient for larger queries:
# find all users with names starting with "m"
async with db.Session() as session:
for user in (await session.stream_scalars(User.select().where(User.name.like('m%')))):
print(user.name)
The results from session.execute()
and session.stream()
are returned as
a list of rows, where each row is a tuple, even if only one result per row was
requested. The scalars()
and stream_scalars()
methods conveniently
return the first result in each row. The scalar()
method returns only the
first object of the first row.
… modify an object stored in a database table?¶
To modify a database object, first retrieve, then modify it within a transaction:
# with an implicit commit
with db.begin() as session:
user = session.get(User, 2)
user.name = 'john'
# with an explicit commit
with db.Session() as session:
user = session.get(User, 2)
user.name = 'john'
session.commit()
With the asynchronous version:
# with an implicit commit
async with db.begin() as session:
user = await session.get(User, 2)
user.name = 'john'
# with an explicit commit
async with db.Session() as session:
user = await session.get(User, 2)
user.name = 'john'
await session.commit()
… delete an object from a database table?¶
To remove an object from the database, use session.delete()
inside a
transaction:
# implicit commit
with db.begin() as session:
user = session.get(User, 2)
session.delete(user)
# explicit commit
with db.Session() as session:
user = session.get(User, 2)
session.delete(user)
session.commit()
If you are using the asynchronous version:
# implicit commit
async with db.begin() as session:
user = await session.get(User, 2)
await session.delete(user)
# explicit commit
async with db.Session() as session:
user = await session.get(User, 2)
await session.delete(user)
await session.commit()
… run an arbitrary SQL statement on the database?¶
Use session.execute()
along with text()
:
from sqlalchemy import text
with db.Session() as session:
sql = text('select * from user;')
results = session.execute(sql).all()
With the asynchronous version:
async with db.Session() as session:
sql = text('select * from user;')
results = (await session.execute(sql)).all()
The asynchronous version also supports streaming the results of a raw SQL statement:
async with db.session() as session:
sql = text('select * from user;')
async for row in await session.stream(sql):
print(row)
One advantage of using the text()
function to create SQL statements is that
it supports binding parameters, which is useful in preventing SQL injection
attacks:
with db.Session() as session:
sql = text('select * from user where user.id = :user_id;')
results = session.execute(sql, params={'user_id': 5}).all()