The most efficient way to update detached objects efficiently as far as I know is to use update() to create UPDATE sql calls which as you saw can be very repetitive and wasteful in its own way, ie. how can you tell what has changed?
You could possibly detach and clone the original db model and then try to determine the diff yourself with the changes after your service makes changes and then replay those changes onto the db model but you are then just re-creating SQLAlchemy's unit of work pattern yourself and only for the simplest case possible and you are probably going to have a bad time...
Luckily in this case SQLAlchemy + the database model is providing most of the Data Access Layer as I understand it:
In software, a data access object (DAO) is a pattern that provides an
abstract interface to some type of database or other persistence
mechanism. By mapping application calls to the persistence layer, the
DAO provides data operations without exposing database details.
SEE: Data access object
As you make changes to database objects within the session SQLAlchemy can track those changes and then perform the appropriate updates to the database. For example setting user.verified = True in my example will automatically submit an UPDATE SQL statement to the database when commit() is called.
This won't always be the most efficient but usually it is fine. If you need to update many rows with complicated conditions then you can drop down to SQL if needed using update() or insert() and building statements in Python.
You can also set echo=True on your engine to see what changes produce what SQL.
Finally sqlalchemy is meant to be used with an understanding of SQL. You can never fully abstract the fact that you are using a database unless you want terrible performance or massive complexity. It sort of provides the best of both worlds, most of the time you don't need to know you are but you can access all the database features when you need them.
Example
I also usually have a service class that provides some more common data access functions, in this example UserService.
Other business logic is placed into various applicable services like AuthService in this example. Sometimes these services will need direct access to the database session but sometimes they will just work directly on the DAO, ie. User, without knowing there is a database at all.
Whether it is via a commandline or a web app I set up a db session, then combine the services to handle the given request and finally commit and close out the session. Hard to replicate a full flow of control here so I just tried to approximate some common tasks.
import os
from dataclasses import dataclass, fields
from sqlalchemy import (
Column,
Integer,
String,
BigInteger,
create_engine,
ForeignKey,
Boolean,
)
from sqlalchemy.sql import (
func,
select,
insert,
text,
)
from sqlalchemy.orm import (
DeclarativeBase,
Session,
relationship
)
from sqlalchemy.schema import MetaData, CreateSchema
def get_engine(env):
return create_engine(f"postgresql+psycopg2://{env['DB_USER']}:{env['DB_PASSWORD']}@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}", echo=True)
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
email = Column(String, nullable=False)
verified = Column(Boolean(), nullable=False, server_default=text('false'), default=False)
verify_token = Column(String, servier_default=text('null'), default=None, nullable=True)
def run(conn):
# After signup maybe we set verify token and send email with link ...
with Session(conn) as db:
user_service = UserService(db=db)
auth_service = AuthService()
u1 = user_service.get_user_by_email('[email protected]')
verify_token = auth_service.set_verify_token(u1)
db.commit()
# Later on after following email link or something better...
with Session(conn) as db:
user_service = UserService(db=db)
auth_service = AuthService()
u1 = user_service.get_user_by_email('[email protected]')
if auth_service.verify_user(u1, verify_token):
print("Verified!")
else:
print("Failed to verified!")
db.commit()
# On a subsequent login we can check if verified
with Session(conn) as db:
user_service = UserService(db=db)
auth_service = AuthService()
u1 = user_service.get_user_by_email('[email protected]')
assert auth_service.is_verified(u1)
class UserService:
""" Handle some common data access functions. """
def __init__(self, db):
self.db = db
def get_user_by_email(self, user_email):
return self.db.scalars(select(User).where(User.email == user_email)).first()
class AuthService:
""" Our business logic goes here. """
def verify_user(self, user, supplied_verify_token):
was_verified = False
if user.verify_token == supplied_verify_token:
user.verified = True
user.verify_token = None
was_verified = True
return was_verified
def set_verify_token(self, user):
user.verify_token = 'MADEUP'
return user.verify_token
def is_verified(self, user):
return user.verified
def populate(conn):
# Make some fake users.
with Session(conn) as session:
u1 = User(name="user1", email="[email protected]")
session.add(u1)
u2 = User(name="user2", email="[email protected]")
session.add(u2)
session.commit()
def main():
engine = get_engine(os.environ)
with engine.begin() as conn:
Base.metadata.create_all(conn)
populate(conn)
run(conn)
if __name__ == '__main__':
main()
Once you get a feel for how things are working and if you have a large project with many many services stringing them all together all the time is not fun and you probably want to use some sort of dependency injection system and create the services with factories not by calling the constructors as I have done in this example.
