Sqlalchemy документация bd

Теги: sqlalchemy  python  bd 

Схема работы

туторилалы

Запуск приложения

Используется объект engine. Engine реализует connection pool. Engine - это глобальный объект, который формируется и конфигурируется только один раз для части сервера, общающегося с ДБ.

>>> from sqlalchemy import create_engine
>>> engine = create_engine("sqlite+pysqlite:///:memory:", echo=True, future=True)

ДБ url задает три вещи: диалект, параметры доступа и имя конкретной БД

dialect+driver://username:password@host:port/database

Пример для [sqlite]

Работа с транзакциями и БД

connect()

>>> from sqlalchemy import text

>>> with engine.connect() as conn:
...     result = conn.execute(text("select 'hello world'"))
...     print(result.all())

BEGIN (implicit)
select 'hello world'
[...] ()
[('hello world',)]
ROLLBACK

commit()

# "commit as you go"
>>> with engine.connect() as conn:
...     conn.execute(text("CREATE TABLE some_table (x int, y int)"))
...     conn.execute(
...         text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
...         [{"x": 1, "y": 1}, {"x": 2, "y": 4}]
...     )
...     conn.commit()

BEGIN (implicit)
CREATE TABLE some_table (x int, y int)
[...] ()
<sqlalchemy.engine.cursor.CursorResult object at 0x...>
INSERT INTO some_table (x, y) VALUES (?, ?)
[...] ((1, 1), (2, 4))
<sqlalchemy.engine.cursor.CursorResult object at 0x...>
COMMIT

Другой метод begin() возвращает коннект с уже открытым комитом

# "begin once"
>>> with engine.begin() as conn:
...     conn.execute(
...         text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
...         [{"x": 6, "y": 8}, {"x": 9, "y": 10}]
...     )

BEGIN (implicit)
INSERT INTO some_table (x, y) VALUES (?, ?)
[...] ((6, 8), (9, 10))
<sqlalchemy.engine.cursor.CursorResult object at 0x...>
COMMIT

Извлечение состояний БД напрямую

Основной метод execute() В коньюнкции с методом text() он возвращает объект result

Объединение строк

>>> with engine.connect() as conn:
...     result = conn.execute(text("SELECT x, y FROM some_table"))
...     for row in result:
...         print(f"x: {row.x}  y: {row.y}")

BEGIN (implicit)
SELECT x, y FROM some_table
[...] ()
x: 1  y: 1
x: 2  y: 4
x: 6  y: 8
x: 9  y: 10
ROLLBACK

Отправка параметров

Построение параметров из полученного ранее представления бд

Извлечение представления при помощи ОРМ-сессии

Используется объект Session. Сессию есть смысл использовать только с ОРМ, т.к. в других случаях она не дает особых преимуществ.

>>> from sqlalchemy.orm import Session

>>> stmt = text("SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y").bindparams(y=6)
>>> with Session(engine) as session:
...     result = session.execute(stmt)
...     for row in result:
...        print(f"x: {row.x}  y: {row.y}")

BEGIN (implicit)
SELECT x, y FROM some_table WHERE y > ? ORDER BY x, y
[...] (6,)
x: 6  y: 8
x: 9  y: 10
x: 11  y: 12
x: 13  y: 14
ROLLBACK

Работа с метаданными БД

Используется объект database metadata, который в пиотоне реализует представление БД строки-колонки. Базовым элементом метаданных БД является table. Метвдвнные можно инициализировать так:

>>> from sqlalchemy import MetaData
>>> metadata = MetaData()

Типичное решение для большинства приложений - один инстантс метаданных. SGLAlchemy позволяет комбинировать множетво для одного приложения. Следующий шан - создание таблиц:

>>> from sqlalchemy import Table, Column, Integer, String
>>> user_table = Table(
...     "user_account",
...     metadata,
...     Column('id', Integer, primary_key=True),
...     Column('name', String(30)),
...     Column('fullname', String)
... )

Здесь инстанс table представляет отдельную таблицу в БД. Колонка представлена инстансом column. Типы данных, такие как Integer могут добавляться в колонку как с помощью, так и без помощи инствнсов.

>>> user_table.c.name
Column('name', String(length=30), table=<user_account>)

>>> user_table.c.keys()
['id', 'name', 'fullname']

Тут-же задаются ограничение, к примеру primary key или foregn key

>>> from sqlalchemy import ForeignKey
>>> address_table = Table(
...     "address",
...     metadata,
...     Column('id', Integer, primary_key=True),
...     Column('user_id', ForeignKey('user_account.id'), nullable=False),
...     Column('email_address', String, nullable=False)
... )

Определение метаданных с помощью ОРМ

Когда мыф используем ОРМ, процесс декларирования метаданных совмещен с процессом декларирования маппированных классов. Маппированный класс - это любой #python класс, который имеет аттрибуты слинкованные с колонками БД. Это делается с помощью объекта registry

>>> from sqlalchemy.orm import registry
>>> mapper_registry = registry()

>>> mapper_registry.metadata
MetaData()

Теперь, вместо декларпирования таблиц и колонок напрямую, мы можем это сделать в маппируемом классе.

>>> Base = mapper_registry.generate_base()

Есть и комбинированный метод, объединяющий registry и декларацию БД

from sqlalchemy.orm import declarative_base
Base = declarative_base()

Теперь осталось задекларировать маппированный классы.

>>> from sqlalchemy.orm import relationship
>>> class User(Base):
...     __tablename__ = 'user_account'
...
...     id = Column(Integer, primary_key=True)
...     name = Column(String(30))
...     fullname = Column(String)
...
...     addresses = relationship("Address", back_populates="user")
...
...     def __repr__(self):
...        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

>>> class Address(Base):
...     __tablename__ = 'address'
...
...     id = Column(Integer, primary_key=True)
...     email_address = Column(String, nullable=False)
...     user_id = Column(Integer, ForeignKey('user_account.id'))
...
...     user = relationship("User", back_populates="addresses")
...
...     def __repr__(self):
...         return f"Address(id={self.id!r}, email_address={self.email_address!r})"

>>> User.__table__
Table('user_account', MetaData(),
    Column('id', Integer(), table=<user_account>, primary_key=True, nullable=False),
    Column('name', String(length=30), table=<user_account>),
    Column('fullname', String(), table=<user_account>), schema=None)

Несколько нюансов:

• класс автоматически генерирует __init__()
• в данном случае реализован __repr__(). но это не обязательно
• связи между таблицами реализованы двусторонее через relationship()

Такой метод создания метаданных сочетаем и с прямым (без ОРм)

Работа с данными в core режиме

Core режим сфокусирован на извлечении и отправке данных при помощи #sql эзыка

Core insert

insert()

>>> from sqlalchemy import insert
>>> stmt = insert(user_table).values(name='spongebob', fullname="Spongebob Squarepants")

>>> print(stmt)
INSERT INTO user_account (name, fullname) VALUES (:name, :fullname)

Можно скомпилировать

>>> compiled = stmt.compile()
>>> compiled.params
{'name': 'spongebob', 'fullname': 'Spongebob Squarepants'}

execute()

>>> with engine.connect() as conn:
...     result = conn.execute(stmt)
...     conn.commit()

BEGIN (implicit)
INSERT INTO user_account (name, fullname) VALUES (?, ?)
[...] ('spongebob', 'Spongebob Squarepants')
COMMIT

Инсерт генерирует значения автоматически из параметров, переданных в execute()

>>> with engine.connect() as conn:
...     result = conn.execute(
...         insert(user_table),
...         [
...             {"name": "sandy", "fullname": "Sandy Cheeks"},
...             {"name": "patrick", "fullname": "Patrick Star"}
...         ]
...     )
...     conn.commit()

BEGIN (implicit)
INSERT INTO user_account (name, fullname) VALUES (?, ?)
[...] (('sandy', 'Sandy Cheeks'), ('patrick', 'Patrick Star'))
COMMIT

Более реалистичный Пример

>>> from sqlalchemy import select, bindparam
>>> scalar_subquery = (
...     select(user_table.c.id).
...     where(user_table.c.name==bindparam('username')).
...     scalar_subquery()
... )

>>> with engine.connect() as conn:
...     result = conn.execute(
...         insert(address_table).values(user_id=scalar_subquery),
...         [
...             {"username": 'spongebob', "email_address": "spongebob@sqlalchemy.org"},
...             {"username": 'sandy', "email_address": "sandy@sqlalchemy.org"},
...             {"username": 'sandy', "email_address": "sandy@squirrelpower.org"},
...         ]
...     )
...     conn.commit()

BEGIN (implicit)
INSERT INTO address (user_id, email_address) VALUES ((SELECT user_account.id
FROM user_account
WHERE user_account.name = ?), ?)
[...] (('spongebob', 'spongebob@sqlalchemy.org'), ('sandy', 'sandy@sqlalchemy.org'),
('sandy', 'sandy@squirrelpower.org'))
COMMIT

Insert.from_select() описание метода

>>> select_stmt = select(user_table.c.id, user_table.c.name + "@aol.com")
>>> insert_stmt = insert(address_table).from_select(
...     ["user_id", "email_address"], select_stmt
... )
>>> print(insert_stmt)

INSERT INTO address (user_id, email_address)
SELECT user_account.id, user_account.name || :name_1 AS anon_1
FROM user_account

Insert.returning() Описание метода

>>> insert_stmt = insert(address_table).returning(address_table.c.id, address_table.c.email_address)
>>> print(insert_stmt)

INSERT INTO address (id, user_id, email_address)
VALUES (:id, :user_id, :email_address)
RETURNING address.id, address.email_address

Selecting Data

select()

>>> from sqlalchemy import select
>>> stmt = select(user_table).where(user_table.c.name == 'spongebob')
>>> print(stmt)

SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = :name_1

Тоже самое через row

>>> with engine.connect() as conn:
...     for row in conn.execute(stmt):
...         print(row)

BEGIN (implicit)
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = ?
[...] ('spongebob',)
(1, 'spongebob', 'Spongebob Squarepants')
ROLLBACK

Или через ОРМ

>>> stmt = select(User).where(User.name == 'spongebob')
>>> with Session(engine) as session:
...     for row in session.execute(stmt):
...         print(row)

BEGIN (implicit)
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = ?
[...] ('spongebob',)
(User(id=1, name='spongebob', fullname='Spongebob Squarepants'),)
ROLLBACK

Примеры работы с колонками

>>> print(select(user_table))
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account

>>> print(select(user_table.c.name, user_table.c.fullname))
SELECT user_account.name, user_account.fullname
FROM user_account

Ну и так далее - всевозможные типы запросов, делиты и т.д.: core вариант работы с данными

Работа с данными в ОРМ

ОРМ режим сфокусирован на ОРМ модели и оперирует объектом Session(). Чтобы работать в ОРМ режиме предварительно нужно знать:

• Executing with an ORM Session (см. пример выше)
• Defining Table Metadata with the ORM (см.пример выше)
• Selecting ORM Entities and Columns (см. пример выше)

Inserting Rows with the ORM

Вначале мы добавляем данные к инстансу объекта Session, затем используя flush() мы отправляем весь пул в БД.

Инстанс класса представляет строки в БД.

>>> squidward = User(name="squidward", fullname="Squidward Tentacles")
>>> krabs = User(name="ehkrabs", fullname="Eugene H. Krabs")

>>> squidward
User(id=None, name='squidward', fullname='Squidward Tentacles')

Значение None означает, что атрибут не был определен к текущему моменту. Маппер sqlalchemy не поднимает ошибку при неверном определении атрибута или неопределении - вместо этого он всегда подставляет представленное значение, либо None, либо пропускает несуществующий аттрибут.

В настоящий момент в моедли данные замаплены согласно модели User, но еще не ассоциированиы ни с какой базой данных. Их необходимо добавить к сессии.

>>> session = Session(engine)

>>> session.add(squidward)
>>> session.add(krabs)

>>> session.new
IdentitySet([User(id=None, name='squidward', fullname='Squidward Tentacles'), User(id=None, name='ehkrabs', fullname='Eugene H. Krabs')])

Объект Session аккумулирует изменеения во времени до тех пор, пока не будет вызван Session.flush.

>>> session.flush()

BEGIN (implicit)
INSERT INTO user_account (name, fullname) VALUES (?, ?)
[...] ('squidward', 'Squidward Tentacles')
INSERT INTO user_account (name, fullname) VALUES (?, ?)
[...] ('ehkrabs', 'Eugene H. Krabs')

Пока Session.flush мы можем выполнить Session.commit(), Session.rollback(), или Session.close()

ОРМ создаст автоматически primary keyдля строк, добавленных во flush

>>> squidward.id
4
>>> krabs.id
5

Все объекты, добавленные в память, мапятся по identity map и доступны через метод get()

>>> some_squidward = session.get(User, 4)
>>> some_squidward
User(id=4, name='squidward', fullname='Squidward Tentacles')

>>> some_squidward is squidward
True

В завершении изменеения необходимо закомиитить, например так

>>> session.commit()
COMMIT

Updating ORM Objects

В core реализации используется UPDATE/DELETE #sql представления. При использовании ОРМ есть два пути реализовать UPDATE:

• праймари - апдейт делается автоматически в процессе Session
• второй вариант - использование Update оператора с поддержкой ОРМ

Primary

>>> sandy = session.execute(select(User).filter_by(name="sandy")).scalar_one()

BEGIN (implicit)
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = ?
[...] ('sandy',)

>>> sandy
User(id=2, name='sandy', fullname='Sandy Cheeks')

# Вносим изменеения
>>> sandy.fullname = "Sandy Squirrel"

>>> sandy in session.dirty
True

session.dirty - сет всех “грязных” объектов, готовых к отправке изменеений.

Перед каждым следующим селектом происходит autoflush и изменеения отправляются в БД

>>> sandy_fullname = session.execute(
...     select(User.fullname).where(User.id == 2)
... ).scalar_one()

UPDATE user_account SET fullname=? WHERE user_account.id = ?
[...] ('Sandy Squirrel', 2)
SELECT user_account.fullname
FROM user_account
WHERE user_account.id = ?
[...] (2,)

>>> print(sandy_fullname)
Sandy Squirrel

>>> sandy in session.dirty
False

ORM-enabled UPDATE statements

Второй вариант - это вызвать вручную update в ОРМ модели

>>> session.execute(
...     update(User).
...     where(User.name == "sandy").
...     values(fullname="Sandy Squirrel Extraodinaire")
... )

UPDATE user_account SET fullname=? WHERE user_account.name = ?
[...] ('Sandy Squirrel Extraodinaire', 'sandy')

В этом случае используется специальная логика для добавления объекта в сессию.

>>> sandy.fullname
'Sandy Squirrel Extraodinaire'

Deleting ORM Objects

>>> patrick = session.get(User, 3)
>>> session.delete(patrick)

До следующего селекта объект будет доступен в сессии. При селекте будет выполнен автофлаш и объект удалится

>>> session.execute(select(User).where(User.name == "patrick")).first()

SELECT address.id AS address_id, address.email_address AS address_email_address,
address.user_id AS address_user_id
FROM address
WHERE ? = address.user_id
[...] (3,)
DELETE FROM user_account WHERE user_account.id = ?
[...] (3,)
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = ?
[...] ('patrick',)

>>> patrick in session
False

Так-же как и с UPDATE мы можем это сделать вручную

>>> # refresh the target object for demonstration purposes
>>> # only, not needed for the DELETE
SQL>>> squidward = session.get(User, 4)

>>> session.execute(delete(User).where(User.name == "squidward"))

DELETE FROM user_account WHERE user_account.name = ?
[...] ('squidward',)

Rolling Back

>>> session.rollback()
ROLLBACK

>>> sandy.fullname
'Sandy Cheeks'

>>> sandy.__dict__
{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x...>,
 'id': 2, 'name': 'sandy', 'fullname': 'Sandy Cheeks'}

>>> patrick in session
True

>>> session.execute(select(User).where(User.name == 'patrick')).scalar_one() is patrick
True

Роллбек отмотал все изменения в сессии, в т.ч. и в терминах БД.

Closing session

>>> session.close()
ROLLBACK

Происходит закрытие всех коннекшенов и транзакций. Это означает, что если делалось только извлечение данных, совсем не обязательно вызывать rollback() - роллбек будет вызван автоматически. Все объекты извлекаются из сессии - это означает, что у них нет функционального состояния, связанного с БД, их, к примеру, нельзя закомитить или получить какие-то даныне из БД

>>> squidward.name
Traceback (most recent call last):
  ...
sqlalchemy.orm.exc.DetachedInstanceError: Instance <User at 0x...> is not bound to a Session; attribute refresh operation cannot proceed

Чтобы прикрепить объекты к сессии снова, необходимо снова заюзать add()

>>> session.add(squidward)
>>> squidward.name

BEGIN (implicit)
SELECT user_account.id AS user_account_id, user_account.name AS user_account_name, user_account.fullname AS user_account_fullname
FROM user_account
WHERE user_account.id = ?
[...] (4,)
'squidward'

Working with Related Objects

Для установки связей между объектами используется relationship()

from sqlalchemy.orm import relationship
class User(Base):
    __tablename__ = 'user_account'

    # ... Column mappings

    addresses = relationship("Address", back_populates="user")


class Address(Base):
    __tablename__ = 'address'

    # ... Column mappings

    user = relationship("User", back_populates="addresses")

Парсинг и загрузка связей

>>> u1 = User(name='pkrabs', fullname='Pearl Krabs')
>>> u1.addresses
[]

>>> a1 = Address(email_address="pearl.krabs@gmail.com")
>>> u1.addresses.append(a1)

# Появились связанные данные
>>> u1.addresses
[Address(id=None, email_address='pearl.krabs@gmail.com')]

>>> a1.user
User(id=None, name='pkrabs', fullname='Pearl Krabs')

Синхронизацию данных обеспечивает relationship.back_populates параметр. Мы можем связать в отношении один ко многим и многие к одному.

>>> a2 = Address(email_address="pearl@aol.com", user=u1)
>>> u1.addresses
[Address(id=None, email_address='pearl.krabs@gmail.com'), Address(id=None, email_address='pearl@aol.com')]

Добавление в сессию

Каскадно добавляются все связанные объекты.

>>> session.add(u1)
>>> u1 in session
True
>>> a1 in session
True
>>> a2 in session
True

>>> print(u1.id)
None
>>> print(a1.user_id)
None

Так как мы не делали коммит, у нас все еще не присвоены айдишники.

>>> session.commit()

INSERT INTO user_account (name, fullname) VALUES (?, ?)
[...] ('pkrabs', 'Pearl Krabs')
INSERT INTO address (email_address, user_id) VALUES (?, ?)
[...] ('pearl.krabs@gmail.com', 6)
INSERT INTO address (email_address, user_id) VALUES (?, ?)
[...] ('pearl@aol.com', 6)
COMMIT

Загрузка связей

>>> u1.id

BEGIN (implicit)
SELECT user_account.id AS user_account_id, user_account.name AS user_account_name,
user_account.fullname AS user_account_fullname
FROM user_account
WHERE user_account.id = ?
[...] (6,)
6

>>> u1.addresses
SELECT address.id AS address_id, address.email_address AS address_email_address,
address.user_id AS address_user_id
FROM address
WHERE ? = address.user_id
[...] (6,)
[Address(id=4, email_address='pearl.krabs@gmail.com'), Address(id=5, email_address='pearl@aol.com')]

Запросы со связями к БД в ОРМ высокоптимизированы, чтобы не быть затратными. Больше о стратегиях загрузки связанных данных можно узнать тут. [sqlalchemy-loader-strategy]

Использование связанных данных в queries

[sqlalchemy-querying]

• Select.join()
• aliased()
• where()
• all()
• any()

Простейшие лоады со связанными данными:

# many to one equals comparison
>>> print(select(Address).where(Address.user == u1))
SELECT address.id, address.email_address, address.user_id
FROM address
WHERE :param_1 = address.user_id

# many to one not equals comparison
>>> print(select(Address).where(Address.user != u1))
SELECT address.id, address.email_address, address.user_id
FROM address
WHERE address.user_id != :user_id_1 OR address.user_id IS NULL

# object is contained in a one-to-many collection
>>> print(select(User).where(User.addresses.contains(a1)))
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.id = :param_1

# An object has a particular parent from a one-to-many perspective
>>> from sqlalchemy.orm import with_parent
>>> print(select(Address).where(with_parent(u1, User.addresses)))
SELECT address.id, address.email_address, address.user_id
FROM address
WHERE :param_1 = address.user_id

100500 дополнительных материалов по ОРМ колонки и типы данных коннекшен и работа с in-memory

Утилиты для sqlalchemy: sqlalchemy_utils. Не пашет с sqlalchemy-1.4. Сурс

You have installed SQLAlchemy 1.4.0. SQLAlchemy-utils is currently not compatible with SQLAlchemy > 1.4.0. The solution is to downgrade SQLAlchemy to 1.3.23.

• [sqlalchemy-querying]
• [fatsapi-sql-orm-example]
• [sqlite]
• [alembic] - тулза для миграций для sqlalchemy
• [sqlalchemy]