См.также
В момент начала разработки SQLAlchemy способ генерации SQL-запросов не был ясен. Текстовый язык мог быть хорошим кандидатом; это стандартный подход, лежащий в основе таких широко известных инструментов объектно-реляционного отображения, как HQL из состава Hibernate. В случае использования языка программирования Python, однако, был доступен более занимательный вариант: использование объектов и выражений языка Python для генерации древовидных структур представления запросов, причем возможным было даже изменение назначения операторов языка Python с целью использования их для формирования SQL-запросов.
Хотя рассматриваемый инструмент и не был первым инструментом, выполняющим подобные функции, следует упомянуть о библиотеке SQLBuilder из состава SQLObject от Ian Bicking, которая была использована как образец при создании системы работы с объектами языка Python и операторами, используемыми в рамках языка формирования запросов SQLAlchemy. При использовании данного подхода объекты языка Python представляют лексические части SQL-запроса. Методы этих объектов, также как и перегружаемые операторы, позволяют генерировать новые унаследованные от существующих лексические конструкции. Наиболее часто используемым объектом является представляющий столбец объект «Column» - библиотека SQLObject будет представлять такие объекты в рамках класса объектно-реляционного отображения, используя пространство имен с доступом посредством атрибута .q; также в SQLAlchemy объявлен атрибут с именем .c. Этот атрибут .c на сегодняшний день поддерживается и используется для представления элементов основной части, подвергающихся выборке, таких, как объекты, представляющие таблицы и запросы выборки.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 | >>> from sqlalchemy import MetaData, Table, Column, String, Integer
>>> metadata = MetaData()
>>> user_table = Table('user', metadata,
... Column('id', Integer, primary_key=True),
... Column('username', String(50)),
... Column('fullname', String(50))
... )
>>> from sqlalchemy import create_engine
>>> engine = create_engine("sqlite://")
>>> metadata.create_all(engine)
[SQL]: PRAGMA table_info("user")
[SQL]: ()
[SQL]:
CREATE TABLE user (
id INTEGER NOT NULL,
username VARCHAR(50),
fullname VARCHAR(50),
PRIMARY KEY (id)
)
[SQL]: ()
[SQL]: COMMIT |
Каждая колонка в SQAlchemy является частью класса
sqlalchemy.sql.expression.ColumnElement
.
В примере ниже показывается соответствие SQL выражения с Python выражением
сравнения. Такое преобразование возможно при помощи реализации «магического»
Python метода sqlalchemy.sql.expression.ColumnElement.__eq__()
.
1 2 3 4 5 6 7 8 | >>> user_table.c.username
Column('username', String(length=50), table=<user>)
>>>
>>> user_table.c.username == 'ed'
<sqlalchemy.sql.expression.BinaryExpression object at 0x7fb829e60a90>
>>>
>>> str(user_table.c.username == 'ed')
'"user".username = :username_1' |
Комбинация нескольких выражений
1 2 3 4 | >>> print(
... (user_table.c.username == 'ed') | (user_table.c.username == 'jack')
... )
"user".username = :username_1 OR "user".username = :username_2 |
SQL операторы OR и AND соответствуют побитовым операторам в Python |
и
&
или функциям sqlalchemy.sql.expression.or_()
и
sqlalchemy.sql.expression.and_()
.
1 2 3 4 5 6 7 8 9 10 11 12 | >>> from sqlalchemy import and_, or_
>>> print(
... and_(
... user_table.c.fullname == 'ed jones',
... or_(
... user_table.c.username == 'ed',
... user_table.c.username == 'jack'
... )
... )
... )
"user".fullname = :fullname_1 AND ("user".username = :username_1 OR "user".username = :username_2) |
Многие операторы наследуются из класса
sqlalchemy.sql.operators.ColumnOperators
SQLAlchemy оператор | Название оператора | Python оператор |
---|---|---|
sqlalchemy.sql.operators.ColumnOperators.__add__() |
add | + |
sqlalchemy.sql.operators.ColumnOperators.__and__() |
and | & |
sqlalchemy.sql.expression.ColumnElement.__eq__() |
equal | == |
sqlalchemy.sql.operators.ColumnOperators.__ge__() |
greater equal | >= |
sqlalchemy.sql.operators.ColumnOperators.__gt__() |
greater than | > |
sqlalchemy.sql.expression.ColumnElement.__le__() |
less equal | <= |
sqlalchemy.sql.expression.ColumnElement.__lt__() |
less than | < |
sqlalchemy.sql.expression.ColumnElement.__ne__() |
not equal | != |
sqlalchemy.sql.operators.ColumnOperators.__or__() |
or | | |
sqlalchemy.sql.operators.ColumnOperators.in_() |
in | in |
sqlalchemy.sql.operators.ColumnOperators.notin_() |
not in | not in |
1 2 3 4 5 6 7 8 9 10 11 12 | >>> print(user_table.c.id == 5)
"user".id = :id_1
>>> print(user_table.c.id >= 5)
"user".id >= :id_1
>>> print(user_table.c.id > 5)
"user".id > :id_1
>>> print(user_table.c.id <= 5)
"user".id <= :id_1
>>> print(user_table.c.id < 5)
"user".id < :id_1
>>> print(user_table.c.id != 5)
"user".id != :id_1 |
Сравнение с None
преобразуется в SQL конструкцию IS NULL
.
1 2 3 4 | >>> print(user_table.c.id != None)
"user".id IS NOT NULL
>>> print(user_table.c.id == None)
"user".id IS NULL |
1 2 3 4 | >>> print((user_table.c.id == None) | (user_table.c.fullname == 'Vasya'))
"user".id IS NULL OR "user".fullname = :fullname_1
>>> print((user_table.c.id == None) & (user_table.c.fullname == 'Vasya'))
"user".id IS NULL AND "user".fullname = :fullname_1 |
Арифметический оператор сложения
1 2 | >>> print(user_table.c.id + 5)
"user".id + :id_1 |
Python оператор сложения автоматически определяет строки и подставляет SQL
оператор конкатенации ||
1 2 | >>> print(user_table.c.fullname + "some name")
"user".fullname || :fullname_1 |
1 2 3 4 5 | >>> print(user_table.c.username.in_(["wendy", "mary", "ed"]))
"user".username IN (:username_1, :username_2, :username_3)
>>> print(user_table.c.username.notin_(["wendy", "mary", "ed"]))
"user".username NOT IN (:username_1, :username_2, :username_3) |
Скомпилированное выражение является объектом класса
sqlalchemy.sql.compiler.SQLCompiler
Диалекты разных СУБД описаны в модулях:
sqlalchemy.dialects.firebird
sqlalchemy.dialects.mssql
sqlalchemy.dialects.mysql
sqlalchemy.dialects.oracle
sqlalchemy.dialects.postgresql
sqlalchemy.dialects.sqlite
sqlalchemy.dialects.sybase
1 2 3 | >>> from sqlalchemy.dialects import sqlite
>>> print(expression.compile(dialect=sqlite.dialect()))
user.username = ? |
1 2 3 4 5 | >>> expression = user_table.c.username == 'ed'
>>> from sqlalchemy.dialects import mysql
>>> print(expression.compile(dialect=mysql.dialect()))
user.username = %s |
1 2 3 | >>> from sqlalchemy.dialects import postgresql
>>> print(expression.compile(dialect=postgresql.dialect()))
"user".username = %(username_1)s |
1 2 3 | >>> from sqlalchemy.dialects import firebird
>>> print(expression.compile(dialect=firebird.dialect()))
"user".username = :username_1 |
1 2 3 | >>> from sqlalchemy.dialects import mssql
>>> print(expression.compile(dialect=mssql.dialect()))
[user].username = :username_1 |
При компиляции SQL выражения буквенные значения преобразуются в параметры, они
доступны через атрибут sqlalchemy.sql.compiler.SQLCompiler.params
1 2 3 4 | >>> expression = user_table.c.username == 'ed'
>>> compiled = expression.compile()
>>> compiled.params
{u'username_1': 'ed'} |
Параметры извлекаются при выполнении запроса
1 2 3 4 5 6 7 8 | >>> engine.execute(
... user_table.select().where(user_table.c.username == 'ed')
... )
[SQL]: SELECT user.id, user.username, user.fullname
FROM user
WHERE user.username = ?
[SQL]: ('ed',)
<sqlalchemy.engine.result.ResultProxy object at 0x7f3714aec3d0> |
SQL запросы INSERT можно формировать при помощи метода
sqlalchemy.schema.Table.insert()
.
1 2 3 4 5 6 7 | >>> insert_stmt = user_table.insert().values(username='ed', fullname='Ed Jones')
>>> conn = engine.connect()
>>> result = conn.execute(insert_stmt)
[SQL]: INSERT INTO user (username, fullname) VALUES (?, ?)
[SQL]: ('ed', 'Ed Jones')
[SQL]: COMMIT |
Результат выполнения содержит в себе значение primary_key
добавленной
записи в БД.
1 2 | >>> result.inserted_primary_key
[1] |
Запись нескольких строк в таблицу за раз.
1 2 3 4 5 6 7 8 9 | >>> conn.execute(user_table.insert(), [
... {'username': 'jack', 'fullname': 'Jack Burger'},
... {'username': 'wendy', 'fullname': 'Wendy Weathersmith'}
>>> ])
[SQL]: INSERT INTO user (username, fullname) VALUES (?, ?)
[SQL]: (('jack', 'Jack Burger'), ('wendy', 'Wendy Weathersmith'))
[SQL]: COMMIT
<sqlalchemy.engine.result.ResultProxy object at 0x7f3714aec810> |
SQL запросы SELECT
можно формировать при помощи функции
sqlalchemy.sql.expression.select()
.
1 2 3 4 5 6 7 8 9 10 11 12 | >>> from sqlalchemy import select
>>> select_stmt = select([user_table.c.username, user_table.c.fullname]).\
... where(user_table.c.username == 'ed')
>>> result = conn.execute(select_stmt)
>>> for row in result:
... print(row)
[SQL]: SELECT user.username, user.fullname
FROM user
WHERE user.username = ?
[SQL]: ('ed',)
(u'ed', u'Ed Jones') |
Выбор всех полей таблицы.
sqlalchemy.engine.ResultProxy.fetchall()
1 2 3 4 5 6 | >>> select_stmt = select([user_table])
>>> conn.execute(select_stmt).fetchall()
FROM user
[SQL]: ()
[(1, u'ed', u'Ed Jones'), (2, u'jack', u'Jack Burger'), (3, u'wendy', u'Wendy Weathersmith'), (4, u'jack', u'Jack Burger'), (5, u'wendy', u'Wendy Weathersmith')] |
Условие WHERE можно указать как метод класса Select
sqlalchemy.sql.expression.Select.where()
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | >>> select_stmt = select([user_table]).\
... where(
... or_(
... user_table.c.username == 'ed',
... user_table.c.username == 'wendy'
... )
... )
>>> conn.execute(select_stmt).fetchall()
[press return to run code]
[SQL]: SELECT user.id, user.username, user.fullname
FROM user
WHERE user.username = ? OR user.username = ?
[SQL]: ('ed', 'wendy')
[(1, u'ed', u'Ed Jones'), (3, u'wendy', u'Wendy Weathersmith'), (5, u'wendy', u'Wendy Weathersmith')] |
Несколько вызовов метода where
сливаются в одно SQL выражения при помощи
оператора AND
.
1 2 3 4 5 6 7 8 9 | >>> select_stmt = select([user_table]).\
... where(user_table.c.username == 'ed').\
... where(user_table.c.fullname == 'ed jones')
>>> conn.execute(select_stmt).fetchall()
[SQL]: SELECT user.id, user.username, user.fullname
FROM user
WHERE user.username = ? AND user.fullname = ?
[SQL]: ('ed', 'ed jones') |
ORDER BY
соответствует методу
sqlalchemy.sql.expression.Select.order_by()
1 2 3 4 5 6 7 8 | >>> select_stmt = select([user_table]).\
... order_by(user_table.c.username)
>>> print(conn.execute(select_stmt).fetchall())
[SQL]: SELECT user.id, user.username, user.fullname
FROM user ORDER BY user.username
[SQL]: ()
[(1, u'ed', u'Ed Jones'), (2, u'jack', u'Jack Burger'), (4, u'jack', u'Jack Burger'), (3, u'wendy', u'Wendy Weathersmith'), (5, u'wendy', u'Wendy Weathersmith')] |
FOREIGN KEY
соответствует классу sqlalchemy.schema.ForeignKey
.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 | >>> from sqlalchemy import ForeignKey
>>> address_table = Table("address", metadata,
... Column('id', Integer, primary_key=True),
... Column('user_id', Integer, ForeignKey('user.id'),
... nullable=False),
... Column('email_address', String(100), nullable=False)
... )
>>> metadata.create_all(engine)
[SQL]: PRAGMA table_info("user")
[SQL]: ()
[SQL]: PRAGMA table_info("address")
[SQL]: ()
[SQL]:
CREATE TABLE address (
id INTEGER NOT NULL,
user_id INTEGER NOT NULL,
email_address VARCHAR(100) NOT NULL,
PRIMARY KEY (id),
FOREIGN KEY(user_id) REFERENCES user (id)
)
[SQL]: ()
[SQL]: COMMIT
>>> conn.execute(address_table.insert(), [
... {"user_id": 1, "email_address": "ed@ed.com"},
... {"user_id": 1, "email_address": "ed@gmail.com"},
... {"user_id": 2, "email_address": "jack@yahoo.com"},
... {"user_id": 3, "email_address": "wendy@gmail.com"},
>>> ])
[SQL]: INSERT INTO address (user_id, email_address) VALUES (?, ?)
[SQL]: ((1, 'ed@ed.com'), (1, 'ed@gmail.com'), (2, 'jack@yahoo.com'), (3, 'wendy@gmail.com'))
[SQL]: COMMIT
<sqlalchemy.engine.result.ResultProxy object at 0x7f3714b0b9d0> |
Два объекта sqlalchemy.schema.Table
могут быть связанны при помощи
метода sqlalchemy.schema.Table.join()
.
1 2 3 4 5 6 7 8 | >>> join_obj = user_table.join(address_table,
... user_table.c.id == address_table.c.user_id)
>>> print(join_obj)
"user" JOIN address ON "user".id = address.user_id
>>> ForeignKey
<class 'sqlalchemy.schema.ForeignKey'>
>>> |
Условие ON подставляется автоматически
1 2 3 | >>> join_obj = user_table.join(address_table)
>>> print(join_obj)
"user" JOIN address ON "user".id = address.user_id |
Выполнить SQL запрос следующей конструкции SELECT FROM JOIN
можно при
помощи метода sqlalchemy.sql.expression.Select.select_from()
.
1 2 3 4 5 6 7 | >>> select_stmt = select([user_table, address_table]).select_from(join_obj)
>>> conn.execute(select_stmt).fetchall()
[SQL]: SELECT user.id, user.username, user.fullname, address.id, address.user_id, address.email_address
FROM user JOIN address ON user.id = address.user_id
[SQL]: ()
[(1, u'ed', u'Ed Jones', 1, 1, u'ed@ed.com'), (1, u'ed', u'Ed Jones', 2, 1, u'ed@gmail.com'), (2, u'jack', u'Jack Burger', 3, 2, u'jack@yahoo.com'), (3, u'wendy', u'Wendy Weathersmith', 4, 3, u'wendy@gmail.com')] |
1 2 3 4 5 6 7 8 9 10 11 | >>> select_stmt = select([user_table]).where(user_table.c.username == 'ed')
>>> print(
... select([select_stmt.c.username]).
... where(select_stmt.c.username == 'ed')
... )
SELECT username
FROM (SELECT "user".id AS id, "user".username AS username, "user".fullname AS fullname
FROM "user"
WHERE "user".username = :username_1)
WHERE username = :username_2 |
Конструкция AS
добавляется при помощи метода
sqlalchemy.sql.expression.Select.alias()
.
1 2 3 4 5 6 7 8 9 10 | >>> select_alias = select_stmt.alias()
>>> print(
... select([select_alias.c.username]).
... where(select_alias.c.username == 'ed')
... )
SELECT anon_1.username
FROM (SELECT "user".id AS id, "user".username AS username, "user".fullname AS fullname
FROM "user"
WHERE "user".username = :username_1) AS anon_1
WHERE anon_1.username = :username_2 |
Примечание
Более сложный пример
Подзапрос использует дополнительно конструкции GROUP BY
и count()
:
1 2 3 4 5 6 7 8 9 10 | >>> from sqlalchemy import func
>>> address_subq = select([
... address_table.c.user_id,
... func.count(address_table.c.id).label('count')
... ]).\
... group_by(address_table.c.user_id).\
... alias()
>>> print(address_subq)
SELECT address.user_id, count(address.id) AS count
FROM address GROUP BY address.user_id |
Вложенный запрос применяет алиас подзапроса.
1 2 3 4 5 6 7 8 9 10 11 12 13 | >>> username_plus_count = select([
... user_table.c.username,
... address_subq.c.count
... ]).select_from(
... user_table.join(address_subq)
... ).order_by(user_table.c.username)
>>> conn.execute(username_plus_count).fetchall()
[SQL]: SELECT user.username, anon_1.count
FROM user JOIN (SELECT address.user_id AS user_id, count(address.id) AS count
FROM address GROUP BY address.user_id) AS anon_1 ON user.id = anon_1.user_id ORDER BY user.username
[SQL]: ()
[(u'ed', 2), (u'jack', 1), (u'wendy', 1)] |
Скаля́р (от лат. scalaris — ступенчатый) — величина, каждое значение которой может быть выражено одним числом. В математике под «числами» могут подразумеваться элементы произвольного поля, тогда когда в физике имеются в виду действительные или комплексные числа. О функции, принимающей скалярные значения, говорят как о скалярной функции.
—WikiPedia
Скалярный SELECT вернет только одно поле одной строки.
1 2 3 4 5 6 7 8 | >>> address_sel = select([
... func.count(address_table.c.id)
... ]).\
... where(user_table.c.id == address_table.c.user_id)
>>> print(address_sel)
SELECT count(address.id) AS count_1
FROM address, "user"
WHERE "user".id = address.user_id |
Чтобы его вызвать в подзапросе нужно использовать метод
sqlalchemy.sql.expression.Select.as_scalar()
1 2 3 4 5 6 7 8 9 | >>> select_stmt = select([user_table.c.username, address_sel.as_scalar()])
>>> conn.execute(select_stmt).fetchall()
[SQL]: SELECT user.username, (SELECT count(address.id) AS count_1
FROM address
WHERE user.id = address.user_id) AS anon_1
FROM user
[SQL]: ()
[(u'ed', 2), (u'jack', 1), (u'wendy', 1), (u'jack', 0), (u'wendy', 0)] |
sqlalchemy.schema.Table.update()
1 2 3 4 5 6 7 8 | >>> update_stmt = address_table.update().\
... values(email_address="jack@msn.com").\
... where(address_table.c.email_address == "jack@yahoo.com")
>>> result = conn.execute(update_stmt)
[SQL]: UPDATE address SET email_address=? WHERE address.email_address = ?
[SQL]: ('jack@msn.com', 'jack@yahoo.com')
[SQL]: COMMIT |
UPDATE запрос значение которого строится из значения полей текущей записи
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | >>> update_stmt = user_table.update().\
... values(fullname=user_table.c.username +
... " " + user_table.c.fullname)
>>> result = conn.execute(update_stmt)
[SQL]: UPDATE user SET fullname=(user.username || ? || user.fullname)
[SQL]: (' ',)
[SQL]: COMMIT
>>> conn.execute(select([user_table])).fetchall()
[SQL]: SELECT user.id, user.username, user.fullname
FROM user
[SQL]: ()
[(1, u'ed', u'ed Ed Jones'), (2, u'jack', u'jack Jack Burger'), (3, u'wendy', u'wendy Wendy Weathersmith'), (4, u'jack', u'jack Jack Burger'), (5, u'wendy', u'wendy Wendy Weathersmith')] |
sqlalchemy.schema.Table.delete()
1 2 3 4 5 6 7 | >>> delete_stmt = address_table.delete().\
... where(address_table.c.email_address == "ed@ed.com")
>>> result = conn.execute(delete_stmt)
[SQL]: DELETE FROM address WHERE address.email_address = ?
[SQL]: ('ed@ed.com',)
[SQL]: COMMIT |
Количество удаленных строк (применимо и для UPDATE).
1 2 | >>> result.rowcount
1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 | # ## slide::
# ## title:: SQL Expression Language
# We begin with a Table object
from sqlalchemy import MetaData, Table, Column, String, Integer
metadata = MetaData()
user_table = Table('user', metadata, Column('id', Integer,
primary_key=True),
Column('username', String(50)),
Column('fullname', String(50)))
# ## slide:: p
# new SQLite database and generate the table.
from sqlalchemy import create_engine
engine = create_engine("sqlite://")
metadata.create_all(engine)
# ## slide::
# as we saw earlier, Table has a collection of Column objects,
# which we can access via table.c.<columnname>
user_table.c.username
# ## slide::
# Column is part of a class known as "ColumnElement",
# which exhibit custom Python expression behavior.
user_table.c.username == 'ed'
# ## slide:: i
# They become SQL when evaluated as a string.
str(user_table.c.username == 'ed')
# ## slide::
# ColumnElements can be further combined to produce more ColumnElements
print((user_table.c.username == 'ed') | (user_table.c.username == 'jack'))
# ## slide::
# OR and AND are available with |, &, or or_() and and_()
from sqlalchemy import and_, or_
print(and_(user_table.c.fullname == 'ed jones',
or_(user_table.c.username == 'ed',
user_table.c.username == 'jack')))
# ## slide::
# comparison operators
print(user_table.c.id > 5)
# ## slide::
# Compare to None produces IS NULL
print(user_table.c.fullname == None) # noqa
# ## slide::
# "+" might mean "addition"....
print(user_table.c.id + 5)
# ## slide:: i
# ...or might mean "string concatenation"
print(user_table.c.fullname + "some name")
# ## slide::
# an IN
print(user_table.c.username.in_(["wendy", "mary", "ed"]))
# ## slide::
# Expressions produce different strings according to *dialect*
# objects.
expression = user_table.c.username == 'ed'
# ## slide:: i
# MySQL....
from sqlalchemy.dialects import mysql
print(expression.compile(dialect=mysql.dialect()))
# ## slide:: i
# PostgreSQL...
from sqlalchemy.dialects import postgresql
print(expression.compile(dialect=postgresql.dialect()))
# ## slide::
# the Compiled object also converts literal values to "bound"
# parameters.
compiled = expression.compile()
compiled.params
# ## slide::
# The "bound" parameters are extracted when we execute()
engine.execute(user_table.select().where(user_table.c.username == 'ed'))
# ## slide::
# ## title:: Exercises
# Produce these expressions using "user_table.c.fullname",
# "user_table.c.id", and "user_table.c.username":
#
# 1. user.fullname = 'ed'
#
# 2. user.fullname = 'ed' AND user.id > 5
#
# 3. user.username = 'edward' OR (user.fullname = 'ed' AND user.id > 5)
#
# ## slide:: p
# we can insert data using the insert() construct
insert_stmt = user_table.insert().values(username='ed', fullname='Ed Jones')
conn = engine.connect()
result = conn.execute(insert_stmt)
# ## slide:: i
# executing an insert() gives us the "last inserted id"
result.inserted_primary_key
# ## slide:: p
# insert() and other DML can run multiple parameters at once.
conn.execute(user_table.insert(),
[{'username': 'jack',
'fullname': 'Jack Burger'},
{'username': 'wendy',
'fullname': 'Wendy Weathersmith'}])
# ## slide:: p
# select() is used to produce any SELECT statement.
from sqlalchemy import select
select_stmt = select([user_table.c.username, user_table.c.fullname]).\
where(user_table.c.username == 'ed')
result = conn.execute(select_stmt)
for row in result:
print(row)
# ## slide:: p
# select all columns from a table
select_stmt = select([user_table])
conn.execute(select_stmt).fetchall()
# ## slide:: p
# specify a WHERE clause
select_stmt = select([user_table]).\
where(
or_(
user_table.c.username == 'ed',
user_table.c.username == 'wendy'
)
)
conn.execute(select_stmt).fetchall()
# ## slide:: p
# specify multiple WHERE, will be joined by AND
select_stmt = select([user_table]).\
where(user_table.c.username == 'ed').\
where(user_table.c.fullname == 'ed jones')
conn.execute(select_stmt).fetchall()
# ## slide:: p
# ordering is applied using order_by()
select_stmt = select([user_table]).\
order_by(user_table.c.username)
print(conn.execute(select_stmt).fetchall())
# ## slide::
# ## title:: Exercises
# 1. use user_table.insert() and "r = conn.execute()" to emit this
# statement:
#
# INSERT INTO user (username, fullname) VALUES ('dilbert', 'Dilbert Jones')
#
# 2. What is the value of 'user.id' for the above INSERT statement?
#
# 3. Using "select([user_table])", execute this SELECT:
#
# SELECT id, username, fullname FROM user WHERE username = 'wendy' OR
# username = 'dilbert' ORDER BY fullname
#
#
# ## slide:: p
# ## title:: Joins / Foreign Keys
# We create a new table to illustrate multi-table operations
from sqlalchemy import ForeignKey
address_table = Table("address", metadata, Column('id', Integer,
primary_key=True),
Column('user_id', Integer, ForeignKey('user.id'),
nullable=False),
Column('email_address', String(100),
nullable=False))
metadata.create_all(engine)
# ## slide:: p
# data
conn.execute(address_table.insert(),
[{"user_id": 1,
"email_address": "ed@ed.com"},
{"user_id": 1,
"email_address": "ed@gmail.com"},
{"user_id": 2,
"email_address": "jack@yahoo.com"},
{"user_id": 3,
"email_address": "wendy@gmail.com"}, ])
# ## slide::
# two Table objects can be joined using join()
#
# <left>.join(<right>, [<onclause>]).
join_obj = user_table.join(address_table,
user_table.c.id == address_table.c.user_id)
print(join_obj)
# ## slide::
# ForeignKey allows the join() to figure out the ON clause automatically
join_obj = user_table.join(address_table)
print(join_obj)
# ## slide:: pi
# to SELECT from a JOIN, use select_from()
select_stmt = select([user_table, address_table]).select_from(join_obj)
conn.execute(select_stmt).fetchall()
# ## slide::
# the select() object is a "selectable" just like Table.
# it has a .c. attribute also.
select_stmt = select([user_table]).where(user_table.c.username == 'ed')
print(select([select_stmt.c.username]).where(select_stmt.c.username == 'ed'))
# ## slide::
# In SQL, a "subquery" is usually an alias() of a select()
select_alias = select_stmt.alias()
print(select([select_alias.c.username]).where(select_alias.c.username == 'ed'))
# ## slide::
# A subquery against "address" counts addresses per user:
from sqlalchemy import func
address_subq = select([
address_table.c.user_id,
func.count(address_table.c.id).label('count')
]).\
group_by(address_table.c.user_id).\
alias()
print(address_subq)
# ## slide:: i
# we use join() to link the alias() with another select()
username_plus_count = select([
user_table.c.username, address_subq.c.count
]).select_from(user_table.join(address_subq)).order_by(user_table.c.username)
# ## slide:: i
conn.execute(username_plus_count).fetchall()
# ## slide::
# ## title:: Exercises
# Produce this SELECT:
#
# SELECT fullname, email_address FROM user JOIN address
# ON user.id = address.user_id WHERE username='ed'
# ORDER BY email_address
#
# ## slide::
# ## title:: Scalar selects, updates, deletes
# a *scalar select* returns exactly one row and one column
address_sel = select([
func.count(address_table.c.id)
]).\
where(user_table.c.id == address_table.c.user_id)
print(address_sel)
# ## slide:: ip
# scalar selects can be used in column expressions,
# specify it using as_scalar()
select_stmt = select([user_table.c.username, address_sel.as_scalar()])
conn.execute(select_stmt).fetchall()
# ## slide:: p
# to round out INSERT and SELECT, this is an UPDATE
update_stmt = address_table.update().\
values(email_address="jack@msn.com").\
where(address_table.c.email_address == "jack@yahoo.com")
result = conn.execute(update_stmt)
# ## slide:: p
# an UPDATE can also use expressions based on other columns
update_stmt = user_table.update().\
values(fullname=user_table.c.username +
" " + user_table.c.fullname)
result = conn.execute(update_stmt)
# ## slide:: i
conn.execute(select([user_table])).fetchall()
# ## slide:: p
# and this is a DELETE
delete_stmt = address_table.delete().\
where(address_table.c.email_address == "ed@ed.com")
result = conn.execute(delete_stmt)
# ## slide:: i
# UPDATE and DELETE have a "rowcount", number of rows matched
# by the WHERE clause.
result.rowcount
# ## slide::
# ## title:: Exercises
# 1. Execute this UPDATE - keep the "result" that's returned
#
# UPDATE user SET fullname='Ed Jones' where username='ed'
#
# 2. how many rows did the above statement update?
#
# 3. Tricky bonus! Combine update() along with select().as_scalar()
# to execute this UPDATE:
#
# UPDATE user SET fullname=fullname ||
# (select email_address FROM address WHERE user_id=user.id)
# WHERE username IN ('jack', 'wendy')
#
# ## slide:: |