Let’s get started 🚀

🌟Session

To start interact with existed database you must have a connection. I would like to use PostgreSQL in every sqlalchemy related posts

import sqlachemy
engine = sqlalchemy.create_engine(
    'postgresql+psycopg2://{0}:{1}@{2}:{3}/{4}'.format(<user>,<password>, <server>, <port>, <database>),
     pool_size=100, max_overflow=200, client_encoding='utf8', executemany_mode="batch"
)
# If you're running version below 1.4.xx. You will need to change executemany_mode="batch" to use_batch_mode=True

To start interact with command either in raw sql or orm style, you mush have a session

from sqlalchemy.orm import Session, sessionmaker

# Session: If you want to have a session using at global state
# sessionmaker: If a session in your local scope whether in module, class or function

session = Session(bind=engine)

session = sessionmaker(bind=engine)

Some basic function from session instance

FunctionsReferences
session.add(instance_warn=True)https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.add
session.add_all(instances)https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.add_all
session.delete(instance)https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.delete
session.execute(statement, …)https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.execute
session.scalar(statement, …)https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.scalar
session.scalars(statement, …)https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.scalars
session.begin()https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.begin
session.commit()https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.commit
session.close()https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.close
session.rollback()https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.rollback

Diference between Session() and Session.begin()

Use within context manager

with Session(bind=engine)() as session:
    ...
    session.commit()
    # No need to close the session

#or

with sessionmaker(bind=engine)() as session:
    ...
    session.commit()
    # No need to close the session

with Session(bind=engine).begin() as session:
    ...
    session.commit()
    # Need to close the session
    session.close()

#or

with sessionmaker(bind=engine).begin() as session:
    ...
    session.commit()
    # Need to close the session
    session.close()

🌟Connection

# runs a transaction
with engine.begin() as connection:
    r1 = connection.execute(statement) # see Statement section below

🌟Statement

Interact with database object with ORM style of contruction

Suppose you have two classes User, Address. Here are some of examples to show you how to contruct a statement

Select individual attributes

from sqlalchemy import select
stmt = select(User.name, User.age)
# Run stmt with session.execute() will be returned a list of tuple of values
results = session.execute(stmt)

Select all attributes

from sqlalchemy import select
stmt = select(User)
# Run stmt with session.execute() will be returned a list of tuple of values
results = session.execute(stmt)
# Run stmt with session.scalars() will be returned a list of User objects
results = session.scalars(stmt)

Join

Example 1

from sqlalchemy import select
stmt = select(User.name, User.age, Address.line).join(
    Address, Address.user_id == User.id
)
# Run stmt with session.execute() will be returned a list of tuple of values
results = session.execute(stmt)

Join with subquery

Select a result from address with a condition email is equal my-email@aol.com

And join User table with selected result set

from sqlalchemy import select, join
subq = (
    select(Address).where(Address.email_address == 'my-email@aol.com').subquery()
)
stmt = select(User).join(subq, User.id == subq.user_id)
# Run stmt with session.execute() will be returned a list of tuple of values
results = session.execute(stmt)
# Run stmt with session.scalars() will be returned a list of Address objects
results = session.scalars(stmt)

Update

Update column fullname = the Start if user’s name is Shawn

from sqlalchemy import update
stmt = (
    update(User).where(User.name == 'Shawn').
    values(fullname='the Star')
)
session.execute(stmt)

We can use bindparam to reuse the statement in executemany context

from sqlalchemy import update, bindparam
stmt = (
    update(User).where(User.name == bindparam('oldname')).values(name=bindparam('newname'))
)
# runs a transaction
with engine.begin() as connection:
    r1 = connection.execute(
        stmt,
        [
            {"oldname": "shawn", "newname": "sang"}
        ]
     )

2 Comments

Leave a Reply

Your email address will not be published.