摘要：可以看作是很多對(duì)象的集合，還有一些關(guān)于的信息。相關(guān)類定義在基礎(chǔ)的模塊中，比如最常用的三個(gè)它也支持同時(shí)定義多個(gè)形成聯(lián)合主鍵。使用獲取單行單列結(jié)果時(shí)需要注意，如果返回多于一行，它會(huì)拋出異常。比如違反唯一性約束等。

SQL Expression Language對(duì)原生SQL語言進(jìn)行了簡單的封裝
兩大模塊SQLAlchemy Core and ORM：

Core：提供執(zhí)行SQL Expression Language的接口

ORM

安裝：SQLAlchemy及相關(guān)數(shù)據(jù)庫驅(qū)動(dòng)
pip install sqlalchemy pymysql

數(shù)據(jù)庫連接字符串格式:請(qǐng)參考這里

mysql://username:password@hostname/database
postgresql://username:password@hostname/database
sqlite:////absolute/path/to/database
oracle://scott:tiger@127.0.0.1:1521/orcl

比如SQLite如下：

from sqlalchemy import create_engine
engine = create_engine("sqlite:///cookies.db")
engine2 = create_engine("sqlite:///:memory:")
engine3 = create_engine("sqlite:////home/cookiemonster/cookies.db")
engine4 = create_engine("sqlite:///c:Userscookiemonstercookies.db")

注意：create_engine函數(shù)返回以一個(gè)engine實(shí)例，但是不會(huì)立即獲取數(shù)據(jù)庫連接，直到在engine上進(jìn)行操作如查詢時(shí)才會(huì)去獲取connection

關(guān)于MySQL空閑連接8小時(shí)自動(dòng)關(guān)閉的解決方案：傳入 pool_recycle=3600參數(shù)

from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://cookiemonster:chocolatechip@mysql01.monster.internal/cookies", pool_recycle=3600)

create_engine其余的一些參數(shù)：

echo：是否log打印執(zhí)行的sql語句及其參數(shù)。默認(rèn)為False

encoding:默認(rèn)utf-8

isolation_level:隔離級(jí)別

pool_recycle：指定連接回收間隔，這對(duì)于MySQL連接的8小時(shí)機(jī)制特別重要。默認(rèn)-1

獲取連接

from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://cookiemonster:chocolatechip" 
"@mysql01.monster.internal/cookies", pool_recycle=3600)
connection = engine.connect()

四種類型集合：
? Generic
? SQL standard
? Vendor specific
? User defined

SQLAlchemy定義了很多generic types以兼容不同數(shù)據(jù)庫。這些類型都定義在sqlalchemy.types模塊中，為了方便也可以從sqlalchemy直接導(dǎo)入這些類型。
類型對(duì)應(yīng)表如下：

如果這些類型不能滿足你，比如有些數(shù)據(jù)庫支持json類型，那么你需要用到sqlalchemy.dialects模塊中對(duì)應(yīng)數(shù)據(jù)庫的類型。比如from sqlalchemy.dialects.postgresql import JSON

Metadata為了快速訪問數(shù)據(jù)庫。可以看作是很多Table對(duì)象的集合，還有一些關(guān)于engin,connection的信息。可以通過MetaData.tables訪問這些表對(duì)象字典
定義表對(duì)象之前需要先實(shí)例化Metadata:

from sqlalchemy import MetaData
metadata = MetaData()

Table對(duì)象構(gòu)建如下：第一個(gè)參數(shù)為名稱，第二個(gè)參數(shù)為Metadata對(duì)象，后續(xù)參數(shù)為Column對(duì)象. Column對(duì)象參數(shù)為,名稱，類型，及其余等

from sqlalchemy import Table, Column, Integer, Numeric, String, ForeignKey
cookies = Table("cookies", metadata,
Column("cookie_id", Integer(), primary_key=True),
Column("cookie_name", String(50), index=True),
Column("cookie_recipe_url", String(255)),
Column("cookie_sku", String(55)),
Column("quantity", Integer()),
Column("unit_cost", Numeric(12, 2))
)

from datetime import datetime
from sqlalchemy import DateTime
users = Table("users", metadata,
Column("user_id", Integer(), primary_key=True),
Column("username", String(15), nullable=False, unique=True),
Column("email_address", String(255), nullable=False),
Column("phone", String(20), nullable=False),
Column("password", String(25), nullable=False),
Column("created_on", DateTime(), default=datetime.now),
Column("updated_on", DateTime(), default=datetime.now, onupdate=datetime.now)

注意：這里default,onupdate屬性是一個(gè)callable對(duì)象而不是直接值，比如datetime.now(),因?yàn)檫@樣的話，就永遠(yuǎn)是這個(gè)值，而不是每個(gè)實(shí)例實(shí)例化、更新時(shí)的時(shí)間了。
比較有用的就是onupdate,每次更新時(shí)都會(huì)調(diào)用該方法或函數(shù)。

鍵和約束(Keys and Constraints)
鍵和約束既可以像上面那樣通過kwargs定義在Column中，也可以在之后通過對(duì)象添加。相關(guān)類定義在基礎(chǔ)的 sqlalchemy模塊中，比如最常用的三個(gè)：
from sqlalchemy import PrimaryKeyConstraint, UniqueConstraint, CheckConstraint

PrimaryKeyConstraint("user_id", name="user_pk"),它也支持同時(shí)定義多個(gè)形成聯(lián)合主鍵。
UniqueConstraint("username", name="uix_username")
CheckConstraint("unit_cost >= 0.00", name="unit_cost_positive")

索引(Index)

from sqlalchemy import Index
Index("ix_cookies_cookie_name", "cookie_name")

這個(gè)定義需要放置在Table構(gòu)造器中。也可以在之后定義，比如
Index("ix_test", mytable.c.cookie_sku, mytable.c.cookie_name))

關(guān)聯(lián)關(guān)系和外鍵約束(Relationships and ForeignKeyConstraints)

from sqlalchemy import ForeignKey
orders = Table("orders", metadata,
Column("order_id", Integer(), primary_key=True),
Column("user_id", ForeignKey("users.user_id")),
Column("shipped", Boolean(), default=False)
)
line_items = Table("line_items", metadata,
Column("line_items_id", Integer(), primary_key=True),
Column("order_id", ForeignKey("orders.order_id")),
Column("cookie_id", ForeignKey("cookies.cookie_id")),
Column("quantity", Integer()),
Column("extended_cost", Numeric(12, 2))
)

注意：這里ForeignKey用的是字符串參數(shù)(這些字符串對(duì)應(yīng)的是數(shù)據(jù)庫中的表名.列名)，而非引用。這樣隔離了模塊間相互依賴
我們也可以使用：
ForeignKeyConstraint(["order_id"], ["orders.order_id"])

創(chuàng)建或持久化表模式(Persisting the Tables)
通過示例代碼我們知道所有的Table定義，以及額外的模式定義都會(huì)與一個(gè)metadata對(duì)象關(guān)聯(lián)。我們可以通過這個(gè)metadata對(duì)象來創(chuàng)建表：

metadata.create_all(engine)

注意：默認(rèn)情況下create_all不會(huì)重新創(chuàng)建已有表，所以它可以安全地多次調(diào)用，而且也非常友好地與數(shù)據(jù)庫遷移庫如Ablembic集成而不需要你進(jìn)行額外手動(dòng)編碼。

本節(jié)代碼完整如下：

from datetime import datetime
from sqlalchemy import (MetaData, Table, Column, Integer, Numeric, String,
DateTime, ForeignKey, create_engine)
metadata = MetaData()
cookies = Table("cookies", metadata,
Column("cookie_id", Integer(), primary_key=True),
Column("cookie_name", String(50), index=True),
Column("cookie_recipe_url", String(255)),
Column("cookie_sku", String(55)),
Column("quantity", Integer()),
Column("unit_cost", Numeric(12, 2))
)
users = Table("users", metadata,
Column("user_id", Integer(), primary_key=True),
Column("customer_number", Integer(), autoincrement=True),
Column("username", String(15), nullable=False, unique=True),
Column("email_address", String(255), nullable=False),
Column("phone", String(20), nullable=False),
Column("password", String(25), nullable=False),
Column("created_on", DateTime(), default=datetime.now),
Column("updated_on", DateTime(), default=datetime.now, onupdate=datetime.now)
)
orders = Table("orders", metadata,
Column("order_id", Integer(), primary_key=True),
Column("user_id", ForeignKey("users.user_id"))
)
line_items = Table("line_items", metadata,
Column("line_items_id", Integer(), primary_key=True),
Column("order_id", ForeignKey("orders.order_id")),
Column("cookie_id", ForeignKey("cookies.cookie_id")),
Column("quantity", Integer()),
Column("extended_cost", Numeric(12, 2))
)
engine = create_engine("sqlite:///:memory:")
metadata.create_all(engine)

ins = cookies.insert().values(
cookie_name="chocolate chip",
cookie_recipe_url="http://some.aweso.me/cookie/recipe.html",
cookie_sku="CC01",
quantity="12",
unit_cost="0.50"
)
print(str(ins))

當(dāng)然你也可以這么做：

from sqlalchemy import insert
ins = insert(cookies).values(
cookie_name="chocolate chip",
cookie_recipe_url="http://some.aweso.me/cookie/recipe.html",
cookie_sku="CC01",
quantity="12",
unit_cost="0.50"
)

上述編譯成預(yù)編譯語句如下：

INSERT INTO cookies
(cookie_name, cookie_recipe_url, cookie_sku, quantity, unit_cost)
VALUES
(:cookie_name, :cookie_recipe_url, :cookie_sku, :quantity, :unit_cost)

實(shí)際過程會(huì)是如下ins對(duì)象內(nèi)部會(huì)調(diào)用compile()方法編譯成上述語句，然后將參數(shù)存儲(chǔ)到ins.compile().params字典中。
接下來我們通過前面獲取的connection對(duì)象執(zhí)行statement：

result = connection.execute(ins)

當(dāng)然你也可以這么查詢：

ins = cookies.insert()
result = connection.execute(
ins,
cookie_name="dark chocolate chip",
cookie_recipe_url="http://some.aweso.me/cookie/recipe_dark.html",
cookie_sku="CC02",
quantity="1",
unit_cost="0.75"
)
result.inserted_primary_key

inventory_list = [
{
"cookie_name": "peanut butter",
"cookie_recipe_url": "http://some.aweso.me/cookie/peanut.html",
"cookie_sku": "PB01",
"quantity": "24",
"unit_cost": "0.25"
},
{
"cookie_name": "oatmeal raisin",
"cookie_recipe_url": "http://some.okay.me/cookie/raisin.html",
"cookie_sku": "EWW01",
"quantity": "100",
"unit_cost": "1.00"
}
]
result = connection.execute(ins, inventory_list)

注意：一定要確保所有字典參數(shù)擁有相同的keys

from sqlalchemy.sql import select
s = select([cookies])
rp = connection.execute(s)
results = rp.fetchall()

當(dāng)然我們也可以使用字符串來代替:

s = select("""SELECT cookies.cookie_id, cookies.cookie_name,
cookies.cookie_recipe_url, cookies.cookie_sku, cookies.quantity,
cookies.unit_cost FROM cookies""")

connection.execute返回的rp變量是一個(gè)ResultProxy對(duì)象(它是DBAPI中cursor對(duì)象的封裝)。

我們也可以這樣寫：

from sqlalchemy.sql import select
s = cookies.select()
rp = connection.execute(s)
results = rp.fetchall()

ResultProxy使得查詢結(jié)果可以通過index,name,or Column object訪問列數(shù)據(jù)。例如：

first_row = results[0]
first_row[1] #游標(biāo)列索引從1開始,by index
first_row.cookie_name # by name
first_row[cookies.c.cookie_name] #by Column object.

你也可以迭代ResultProxy,如下：

rp = connection.execute(s)
for record in rp:
print(record.cookie_name)

ResultProxy其余可用來獲取結(jié)果集的方法

first()

fetchone()

fetchall()

scalar():Returns a single value if a query results in a single record with one column.

keys() 獲取列名

關(guān)于選擇ResultProxy上述的方法的建議：
1、使用first()而不是fetchone()來獲取單條記錄，因?yàn)閒etchone()調(diào)用之后仍然保留著打開的connections共后續(xù)使用，如果不小心的話很容易引起問題。
2、使用迭代方式獲取所有結(jié)果，而不是fetchall(),更加省內(nèi)存。
3、使用scalar()獲取單行單列結(jié)果時(shí)需要注意，如果返回多于一行，它會(huì)拋出異常。

控制返回列的數(shù)目

s = select([cookies.c.cookie_name, cookies.c.quantity])
rp = connection.execute(s)
print(rp.keys())
result = rp.first()

排序

s = select([cookies.c.cookie_name, cookies.c.quantity])
s = s.order_by(cookies.c.quantity)
rp = connection.execute(s)
for cookie in rp:
print("{} - {}".format(cookie.quantity, cookie.cookie_name))

#倒序desc
from sqlalchemy import desc
s = select([cookies.c.cookie_name, cookies.c.quantity])
s = s.order_by(desc(cookies.c.quantity))

限制返回結(jié)果集的條數(shù)

s = select([cookies.c.cookie_name, cookies.c.quantity])
s = s.order_by(cookies.c.quantity)
s = s.limit(2)
rp = connection.execute(s)
print([result.cookie_name for result in rp])

在sqlalchemy.sql.func模塊中

#sum
from sqlalchemy.sql import func
s = select([func.sum(cookies.c.quantity)])
rp = connection.execute(s)
print(rp.scalar())

#count
s = select([func.count(cookies.c.cookie_name)])
rp = connection.execute(s)
record = rp.first()
print(record.keys())
print(record.count_1) #字段名是自動(dòng)生成的，_，可以設(shè)置別名的，看下面

#設(shè)置別名
s = select([func.count(cookies.c.cookie_name).label("inventory_count")])
rp = connection.execute(s)
record = rp.first()
print(record.keys())
print(record.inventory_count)

過濾

#where
s = select([cookies]).where(cookies.c.cookie_name == "chocolate chip")
rp = connection.execute(s)
record = rp.first()
print(record.items()) #調(diào)用row對(duì)象的items()方法。

#like
s = select([cookies]).where(cookies.c.cookie_name.like("%chocolate%"))
rp = connection.execute(s)
for record in rp.fetchall():
    print(record.cookie_name)

可以在where中使用的子句元素

between(cleft, cright)

concat(column_two) Concatenate column with column_two

distinct()

in_([list])

is_(None) Find where the column is None (commonly used for Null checks with None)

contains(string) Find where the column has string in it (case-sensitive)

endswith(string) Find where the column ends with string (case-sensitive)

like(string) Find where the column is like string (case-sensitive)

startswith(string) Find where the column begins with string (case-sensitive)

ilike(string) Find where the column is like string (this is not case-sensitive)

當(dāng)然還包括一系列的notxxx方法，比如notin_(),唯一的例外是isnot()

操作符

+，-，*,/,%

==,!=,<,>,<=,>=

AND,OR,NOT,由于python關(guān)鍵字的原因，使用and_(),or_(),not_()來代替

+號(hào)還可以用于字符串拼接：

s = select([cookies.c.cookie_name, "SKU-" + cookies.c.cookie_sku])
for row in connection.execute(s):
print(row)

from sqlalchemy import cast
s = select([cookies.c.cookie_name,
    cast((cookies.c.quantity * cookies.c.unit_cost),
        Numeric(12,2)).label("inv_cost")])
for row in connection.execute(s):
    print("{} - {}".format(row.cookie_name, row.inv_cost))

注意：cast是另外一個(gè)函數(shù)，允許我們進(jìn)行類型轉(zhuǎn)換，上述轉(zhuǎn)換是將數(shù)字轉(zhuǎn)換為貨幣形式，和
print("{} - {:.2f}".format(row.cookie_name, row.inv_cost)).這個(gè)行為一致。

from sqlalchemy import and_, or_, not_
s = select([cookies]).where(
    and_(
        cookies.c.quantity > 23,
        cookies.c.unit_cost < 0.40
    )
)
for row in connection.execute(s):
    print(row.cookie_name)


from sqlalchemy import and_, or_, not_
s = select([cookies]).where(
    or_(
        cookies.c.quantity.between(10, 50),
        cookies.c.cookie_name.contains("chip")
    )
)
for row in connection.execute(s):
    print(row.cookie_name)

from sqlalchemy import update
u = update(cookies).where(cookies.c.cookie_name == "chocolate chip")
u = u.values(quantity=(cookies.c.quantity + 120))
result = connection.execute(u)
print(result.rowcount)
s = select([cookies]).where(cookies.c.cookie_name == "chocolate chip")
result = connection.execute(s).first()
for key in result.keys():
    print("{:>20}: {}".format(key, result[key]))

from sqlalchemy import delete
u = delete(cookies).where(cookies.c.cookie_name == "dark chocolate chip")
result = connection.execute(u)
print(result.rowcount)

s = select([cookies]).where(cookies.c.cookie_name == "dark chocolate chip")
result = connection.execute(s).fetchall()
print(len(result))

join()，outerjoin()函數(shù)，select_from()函數(shù)

columns = [orders.c.order_id, users.c.username, users.c.phone,
           cookies.c.cookie_name, line_items.c.quantity,
           line_items.c.extended_cost]
cookiemon_orders = select(columns)
cookiemon_orders = cookiemon_orders.select_from(orders.join(users).join(
    line_items).join(cookies)).where(users.c.username ==
                                     "cookiemon")
result = connection.execute(cookiemon_orders).fetchall()
for row in result:
    print(row)

最終產(chǎn)生的SQL語句如下：

SELECT orders.order_id, users.username, users.phone, cookies.cookie_name,
line_items.quantity, line_items.extended_cost FROM users JOIN orders ON
users.user_id = orders.user_id JOIN line_items ON orders.order_id =
line_items.order_id JOIN cookies ON cookies.cookie_id = line_items.cookie_id
WHERE users.username = :username_1

outerjoin

columns = [users.c.username, func.count(orders.c.order_id)]
all_orders = select(columns)
all_orders = all_orders.select_from(users.outerjoin(orders))
all_orders = all_orders.group_by(users.c.username)
result = connection.execute(all_orders).fetchall()
for row in result:
    print(row)

表別名函數(shù)alias()

>>> manager = employee_table.alias("mgr")
>>> stmt = select([employee_table.c.name],
            ... and_(employee_table.c.manager_id==manager.c.id,
            ... manager.c.name=="Fred"))
>>> print(stmt)
SELECT employee.name
FROM employee, employee AS mgr
WHERE employee.manager_id = mgr.id AND mgr.name = ?

columns = [users.c.username, func.count(orders.c.order_id)]
all_orders = select(columns)
all_orders = all_orders.select_from(users.outerjoin(orders))
all_orders = all_orders.group_by(users.c.username)
result = connection.execute(all_orders).fetchall()
for row in result:
    print(row)

def get_orders_by_customer(cust_name, shipped=None, details=False):
    columns = [orders.c.order_id, users.c.username, users.c.phone]
    joins = users.join(orders)
    if details:
        columns.extend([cookies.c.cookie_name, line_items.c.quantity,
            line_items.c.extended_cost])
        joins = joins.join(line_items).join(cookies)
    cust_orders = select(columns)
    cust_orders = cust_orders.select_from(joins)

    cust_orders = cust_orders.where(users.c.username == cust_name)
    if shipped is not None:
        cust_orders = cust_orders.where(orders.c.shipped == shipped)
    result = connection.execute(cust_orders).fetchall()
    return result

返回的還是ResultProxy對(duì)象
1、完全采用原始SQL

result = connection.execute("select * from orders").fetchall()
print(result)

2、部分采用原始SQL，text()函數(shù)

from sqlalchemy import text
stmt = select([users]).where(text("username="cookiemon""))
print(connection.execute(stmt).fetchall())

SQLALchemy定義了很多異常。我們通過關(guān)心：AttributeErrors，IntegrityErrors.等
為了進(jìn)行相關(guān)試驗(yàn)與說明，請(qǐng)先執(zhí)行下面這些語句

from datetime import datetime
from sqlalchemy import (MetaData, Table, Column, Integer, Numeric, String,
                        DateTime, ForeignKey, Boolean, create_engine,
                        CheckConstraint)
metadata = MetaData()
cookies = Table("cookies", metadata,
                Column("cookie_id", Integer(), primary_key=True),
                37
                Column("cookie_name", String(50), index=True),
                Column("cookie_recipe_url", String(255)),
                Column("cookie_sku", String(55)),
                Column("quantity", Integer()),
                Column("unit_cost", Numeric(12, 2)),
                CheckConstraint("quantity > 0", name="quantity_positive")
                )
users = Table("users", metadata,
              Column("user_id", Integer(), primary_key=True),
              Column("username", String(15), nullable=False, unique=True),
              Column("email_address", String(255), nullable=False),
              Column("phone", String(20), nullable=False),
              Column("password", String(25), nullable=False),
              Column("created_on", DateTime(), default=datetime.now),
              Column("updated_on", DateTime(),
                     default=datetime.now, onupdate=datetime.now)
              )
orders = Table("orders", metadata,
               Column("order_id", Integer()),
               Column("user_id", ForeignKey("users.user_id")),
               Column("shipped", Boolean(), default=False)
               )
line_items = Table("line_items", metadata,
                   Column("line_items_id", Integer(), primary_key=True),
                   Column("order_id", ForeignKey("orders.order_id")),
                   Column("cookie_id", ForeignKey("cookies.cookie_id")),
                   Column("quantity", Integer()),
                   Column("extended_cost", Numeric(12, 2))
                   )
engine = create_engine("sqlite:///:memory:")
metadata.create_all(engine)
connection = engine.connect()

from sqlalchemy import select, insert
ins = insert(users).values(
username="cookiemon",
email_address="mon@cookie.com",
phone="111-111-1111",
password="password"
)
result = connection.execute(ins)
s = select([users.c.username])
results = connection.execute(s)
for result in results:
print(result.username)
print(result.password) #此處包AttributeError異常

在違反約束的情況下會(huì)出現(xiàn)IntegrityError異常。比如違反唯一性約束等。

s = select([users.c.username])
connection.execute(s).fetchall()
[(u"cookiemon",)]
ins = insert(users).values(
    username="cookiemon",
    email_address="damon@cookie.com",
    phone="111-111-1111",
    password="password"
)
result = connection.execute(ins) #此處報(bào)IntegrityError, UNIQUE constraint failed: users.username
#異常處理
try:
    result = connection.execute(ins)
except IntegrityError as error:
    print(error.orig.message, error.params)

所有的SQLAlchemy異常處理方式都是上面那種思路，通過[SQLAlchemyError](http://docs.sqlal
chemy.org/en/latest/core/exceptions.html)可以獲取到的信息由如下：

orig ：The DBAPI exception object.

params：The parameter list being used when this exception occurred.

statement ：The string SQL statement being invoked when this exception occurred.

from sqlalchemy.exc import IntegrityError


def ship_it(order_id):
    s = select([line_items.c.cookie_id, line_items.c.quantity])
    s = s.where(line_items.c.order_id == order_id)
    transaction = connection.begin() #開啟事務(wù)
    cookies_to_ship = connection.execute(s).fetchall()
    try:
        for cookie in cookies_to_ship:
            u = update(cookies).where(cookies.c.cookie_id == cookie.cookie_id)
            u = u.values(quantity=cookies.c.quantity - cookie.quantity)
            result = connection.execute(u)
        u = update(orders).where(orders.c.order_id == order_id)
        u = u.values(shipped=True)
        result = connection.execute(u)
        print("Shipped order ID: {}".format(order_id))
        transaction.commit() #提交事務(wù)
    except IntegrityError as error:
        transaction.rollback()   #事務(wù)回滾
        print(error)