摘要：你應(yīng)該使用工廠(chǎng)類(lèi)來(lái)創(chuàng)建類(lèi)，因?yàn)檫@確保了配置參數(shù)的正確性。對(duì)象包含創(chuàng)建數(shù)據(jù)庫(kù)連接所需的一切信息，它不會(huì)立即創(chuàng)建連接對(duì)象，而是會(huì)在我們進(jìn)行具體操作時(shí)創(chuàng)建。注意生產(chǎn)環(huán)境不要使用這個(gè)選項(xiàng)。關(guān)于選擇的最佳實(shí)踐使用迭代方式獲取所有值，而不是。

定義ORM類(lèi)的4個(gè)步驟：

繼承declarative_base()函數(shù)返回的類(lèi)

定義__tablename__屬性來(lái)指定表名

定義列屬性

定義至少一個(gè)主鍵

from sqlalchemy import Table, Column, Integer, Numeric, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Cookie(Base):
    __tablename__ = "cookies"
    
    cookie_id = Column(Integer(), primary_key=True)
    cookie_name = Column(String(50), index=True)
    cookie_recipe_url = Column(String(255))
    cookie_sku = Column(String(55))
    quantity = Column(Integer())
    unit_cost = Column(Numeric(12, 2))

你可以查看Cookie類(lèi)的__table__屬性：如下

>>> Cookie.__table__
Table("cookies", MetaData(bind=None),
    Column("cookie_id", Integer(), table=, primary_key=True,
        nullable=False),
    Column("cookie_name", String(length=50), table=),
    Column("cookie_recipe_url", String(length=255), table=),
    Column("cookie_sku", String(length=15), table=),
    Column("quantity", Integer(), table=),
    Column("unit_cost", Numeric(precision=12, scale=2),
        table=), schema=None)

class SomeDataClass(Base):
__tablename__ = "somedatatable"
__table_args__ = (ForeignKeyConstraint(["id"], ["other_table.id"]),
                  CheckConstraint(unit_cost >= 0.00",
                                  name="unit_cost_positive"))

from sqlalchemy import ForeignKey, Boolean
from sqlalchemy.orm import relationship, backref

class Order(Base):
    __tablename__ = "orders"
    order_id = Column(Integer(), primary_key=True)
    #定義外鍵
    user_id = Column(Integer(), ForeignKey("users.user_id"))
    shipped = Column(Boolean(), default=False)
    #定義one-to-many關(guān)系
    user = relationship("User", backref=backref("orders", order_by=order_id))

class LineItem(Base):
    __tablename__ = "line_items"
    line_item_id = Column(Integer(), primary_key=True)
    order_id = Column(Integer(), ForeignKey("orders.order_id"))
    cookie_id = Column(Integer(), ForeignKey("cookies.cookie_id"))
    quantity = Column(Integer())
    extended_cost = Column(Numeric(12, 2))
    order = relationship("Order", backref=backref("line_items",
                                                  order_by=line_item_id))
    #定義one-to-one關(guān)系，uselist=False
    cookie = relationship("Cookie", uselist=False)

from sqlalchemy import create_engine
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine) #這個(gè)Base是前面的Base = declarative_base()

session對(duì)象負(fù)責(zé)與數(shù)據(jù)庫(kù)交互，封裝了來(lái)自engine的connection，transaction.session中的事物會(huì)一直打開(kāi)，除非調(diào)用session的commit()或rollback()方法，或close(),remove()方法。
你應(yīng)該使用sessionmaker工廠(chǎng)類(lèi)來(lái)創(chuàng)建Session類(lèi)，因?yàn)檫@確保了配置參數(shù)的正確性。一個(gè)應(yīng)用應(yīng)該只調(diào)用sessionmaker一次。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine("sqlite:///:memory:")
Session = sessionmaker(bind=engine)
session = Session()

session對(duì)象包含創(chuàng)建數(shù)據(jù)庫(kù)連接所需的一切信息，它不會(huì)立即創(chuàng)建連接對(duì)象，而是會(huì)在我們進(jìn)行具體操作時(shí)創(chuàng)建。

cc_cookie = Cookie(cookie_name="chocolate chip",
cookie_recipe_url="http://some.aweso.me/cookie/recipe.html",
cookie_sku="CC01",
quantity=12,
unit_cost=0.50)
session.add(cc_cookie)
session.commit()

當(dāng)我們調(diào)用add()的時(shí)候，它不會(huì)在數(shù)據(jù)庫(kù)執(zhí)行insert操作，而當(dāng)我們調(diào)用commit()的時(shí)候，將會(huì)發(fā)生如下步驟：

#start a transaction.
INFO:sqlalchemy.engine.base.Engine:BEGIN (implicit)
#Insert the record into the database
INFO:sqlalchemy.engine.base.Engine:INSERT INTO cookies (cookie_name,
cookie_recipe_url, cookie_sku, quantity, unit_cost) VALUES (?, ?, ?, ?, ?)

#The values for the insert.
INFO:sqlalchemy.engine.base.Engine:("chocolate chip",
"http://some.aweso.me/cookie/recipe.html", "CC01", 12, 0.5) 
#Commit the transaction.
INFO:sqlalchemy.engine.base.Engine:COMMIT

如果你想打印這些細(xì)節(jié)信息，你可以傳遞echo=True到create_engine函數(shù)中。注意生產(chǎn)環(huán)境不要使用這個(gè)選項(xiàng)。

文檔地址
Session.bulk_save_objects()，Session.bulk_update_mappings()

c1 = Cookie(cookie_name="peanut butter",
            cookie_recipe_url="http://some.aweso.me/cookie/peanut.html",
            cookie_sku="PB01",
            quantity=24,
            unit_cost=0.25)
c2 = Cookie(cookie_name="oatmeal raisin",
            cookie_recipe_url="http://some.okay.me/cookie/raisin.html",
            cookie_sku="EWW01",
            quantity=100,
            unit_cost=1.00)
session.bulk_save_objects([c1, c2])
session.commit()
print(c1.cookie_id)

除了bulk_save_objects，還有Session.bulk_update_mappings(), 如下：
它允許我們通過(guò)字典列表來(lái)進(jìn)行插入

s.bulk_insert_mappings(User,
  [dict(name="u1"), dict(name="u2"), dict(name="u3")]
)

對(duì)于批量更新，還有個(gè)Session.bulk_update_mappings()

cookies = session.query(Cookie).all()
print(cookies)

#使用迭代方式
for cookie in session.query(Cookie):
    print(cookie)

其余方法：

all()

first()

one():如果有多條結(jié)果，會(huì)拋出異常。

scalar()

關(guān)于選擇的最佳實(shí)踐：
1、使用迭代方式獲取所有值，而不是all()。內(nèi)存友好
2、使用first()獲取單條數(shù)據(jù)，而不是one(),scalar()
3、盡量不要使用scalar()

控制查詢(xún)的列數(shù)目

print(session.query(Cookie.cookie_name, Cookie.quantity).first())

排序

for cookie in session.query(Cookie).order_by(Cookie.quantity):
    print("{:3} - {}".format(cookie.quantity, cookie.cookie_name))
    
from sqlalchemy import desc
for cookie in session.query(Cookie).order_by(desc(Cookie.quantity)):
    print("{:3} - {}".format(cookie.quantity, cookie.cookie_name))

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

query = session.query(Cookie).order_by(Cookie.quantity).limit(2)
print([result.cookie_name for result in query])

內(nèi)置SQL函數(shù)與別名

from sqlalchemy import func
inv_count = session.query(func.sum(Cookie.quantity)).scalar()
print(inv_count)

rec_count = session.query(func.count(Cookie.cookie_name)).first()
print(rec_count) #(5,) 得到的是一個(gè)元組，而不是像scalar()那樣得到單個(gè)值

#別名
rec_count = session.query(func.count(Cookie.cookie_name) 
.label("inventory_count")).first()
print(rec_count.keys())
print(rec_count.inventory_count)

record = session.query(Cookie).filter(Cookie.cookie_name == "chocolate chip").first()
print(record)
record = session.query(Cookie).filter_by(cookie_name="chocolate chip").first()
print(record)

注意：filter與filter_by的區(qū)別

query = session.query(Cookie).filter(Cookie.cookie_name.like("%chocolate%"))
for record in query:
    print(record.cookie_name)

操作符

+，-，*,/,%

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

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

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

results = session.query(Cookie.cookie_name, "SKU-" + Cookie.cookie_sku).all()
for row in results:
    print(row)
    
from sqlalchemy import and_, or_, not_
query = session.query(Cookie).filter(or_(
    Cookie.quantity.between(10, 50),
    Cookie.cookie_name.contains("chip")
    )
)
for result in query:
    print(result.cookie_name)

query = session.query(Cookie)
cc_cookie = query.filter(Cookie.cookie_name == "chocolate chip").first()
cc_cookie.quantity = cc_cookie.quantity + 120
session.commit()
print(cc_cookie.quantity)


#通過(guò)字典方式更新
query = session.query(Cookie)
query = query.filter(Cookie.cookie_name == "chocolate chip")
query.update({Cookie.quantity: Cookie.quantity - 20})
cc_cookie = query.first()
print(cc_cookie.quantity)

query = session.query(Cookie)
query = query.filter(Cookie.cookie_name == "dark chocolate chip")
dcc_cookie = query.one()
session.delete(dcc_cookie)
session.commit()
dcc_cookie = query.first()
print(dcc_cookie)

#或者這樣
query = session.query(Cookie)
query = query.filter(Cookie.cookie_name == "molasses")
query.delete()

添加關(guān)聯(lián)對(duì)象

o1 = Order()
o1.user = cookiemon
session.add(o1)
cc = session.query(Cookie).filter(Cookie.cookie_name ==
                                  "chocolate chip").one()
line1 = LineItem(cookie=cc, quantity=2, extended_cost=1.00)
pb = session.query(Cookie).filter(Cookie.cookie_name ==
                                  "peanut butter").one()
line2 = LineItem(quantity=12, extended_cost=3.00)
line2.cookie = pb
line2.order = o1
o1.line_items.append(line1)
o1.line_items.append(line2)
session.commit()

query = session.query(Order.order_id, User.username, User.phone,
                      Cookie.cookie_name, LineItem.quantity,
                      LineItem.extended_cost)
query = query.join(User).join(LineItem).join(Cookie)
results = query.filter(User.username == "cookiemon").all()
print(results)


query = session.query(User.username, func.count(Order.order_id))
query = query.outerjoin(Order).group_by(User.username)
for row in query:
    print(row)

class Employee(Base):
    __tablename__ = "employees"
    id = Column(Integer(), primary_key=True)
    manager_id = Column(Integer(), ForeignKey("employees.id"))
    name = Column(String(255), nullable=False)
    manager = relationship("Employee", backref=backref("reports"),
                           remote_side=[id])
Base.metadata.create_all(engine)

注：使用remote_side來(lái)定義自關(guān)聯(lián)的多對(duì)一關(guān)系

marsha = Employee(name="Marsha")
fred = Employee(name="Fred")
marsha.reports.append(fred)
session.add(marsha)
session.commit()

for report in marsha.reports:
    print(report.name)

query = session.query(User.username, func.count(Order.order_id))
query = query.outerjoin(Order).group_by(User.username)
for row in query:
    print(row)

def get_orders_by_customer(cust_name):
    query = session.query(Order.order_id, User.username, User.phone,
                          Cookie.cookie_name, LineItem.quantity,
                          LineItem.extended_cost)
    query = query.join(User).join(LineItem).join(Cookie)
    results = query.filter(User.username == cust_name).all()
    return results
get_orders_by_customer("cakeeater")

session.execute("select * from User")
session.execute("insert into User(name, age) values("bomo", 13)")
session.execute("insert into User(name, age) values(:name, :age)", {"name": "bomo", "age":12})

建議使用text()來(lái)執(zhí)行部分SQL查詢(xún)

from sqlalchemy import text
query = session.query(User).filter(text("username="cookiemon""))
print(query.all())

[User(username="cookiemon", email_address="[email protected]",
      phone="111-111-1111", password="password")]

Session狀態(tài)：

Transient：實(shí)例不在session和數(shù)據(jù)庫(kù)中。

Pending：對(duì)象通過(guò)add()方法被添加到session當(dāng)中，但是并沒(méi)有flushed或者committed

Persistent:對(duì)象處于session中，同時(shí)在數(shù)據(jù)庫(kù)中有對(duì)應(yīng)的記錄

Detached:實(shí)例不在session中，但是數(shù)據(jù)庫(kù)中有相關(guān)記錄

那么如何查看實(shí)例狀態(tài)呢？可以通過(guò)SQLAlchemy的inspect()方法來(lái)查看,

cc_cookie = Cookie("chocolate chip",
                   "http://some.aweso.me/cookie/recipe.html",
                   "CC01", 12, 0.50)
from sqlalchemy import inspect
insp = inspect(cc_cookie)
for state in ["transient", "pending", "persistent", "detached"]:
    print("{:>10}: {}".format(state, getattr(insp, state)))

輸出：
transient: True
pending: False
persistent: False
detached: False
實(shí)際上，你應(yīng)該使用insp.transient, insp.pending, insp.persistent, and insp.detached來(lái)獲取某一個(gè)狀態(tài)。

如果要將一個(gè)實(shí)例變?yōu)閐etached狀態(tài)，可以調(diào)用session的expunge()方法

session.expunge(cc_cookie)

查看改變歷史

for attr, attr_state in insp.attrs.items():
    if attr_state.history.has_changes():
        print("{}: {}".format(attr, attr_state.value))
        print("History: {}
".format(attr_state.history))

文檔
我們關(guān)系的主要有兩個(gè)MultipleResultsFound，DetachedInstanceError.

from sqlalchemy.orm.exc import MultipleResultsFound
try:
    results = session.query(Cookie).one()
except MultipleResultsFound as error:
    print("We found too many cookies... is that even possible?")