Домой Edit me on GitHub

2020-12-05

Каналы передачи данных | Сетевое программирование | Базы данных | Основы Веб-программирования

SQL выражения

В момент начала разработки 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

Функции OR и AND

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

Соответствие магических методов Python и переопределенных методов в SQLAlchemy
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

Операторы AND и OR

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

Операторы IN и NOT IN

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)

Компиляция SQL выражений

Скомпилированное выражение является объектом класса 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

SQLite

1
2
3
>>> from sqlalchemy.dialects import sqlite
>>> print(expression.compile(dialect=sqlite.dialect()))
user.username = ?

MySQL

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

PostgreSQL

1
2
3
>>> from sqlalchemy.dialects import postgresql
>>> print(expression.compile(dialect=postgresql.dialect()))
"user".username = %(username_1)s

Firebird

1
2
3
>>> from sqlalchemy.dialects import firebird
>>> print(expression.compile(dialect=firebird.dialect()))
"user".username = :username_1

MSSQL

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>

INSERT

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>

SELECT

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

Условие 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

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

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>

JOIN

Два объекта 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)

Конструкция 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)]

UPDATE

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')]

DELETE

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

Полный пример

2.sqlalchemy/3.sql_expression.py
  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::
Previous: Метаданные (metadata) Next: ORM (объектно-реляционное отображение)