sqlalchemy 的 ORM 方式使用示例

知乎： 使用 sqlalchemy 的 orm 方式操作數據庫是一種怎樣的體驗？

答： 酸爽！

 

本文基於：win10 + python3.4 + sqlAlchemy 1.0.13

先看一個圖（來源）：

這是 sqlalchemy 的層級圖。不難發現，其中 orm 是最頂級的封裝。

 

ORM 基本操作步驟如下：

1. 建立連接

from sqlalchemy import create_engine engine = create_engine('sqlite:///:memory:', echo=True)

 

2. 建立會話

from sqlalchemy.orm import Session session = Session(engine)

 

3. 聲明基類

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

 

4. 定義表（繼承基類）

from sqlalchemy import Column, Integer, ForeignKey
from sqlalchemy.orm import relationship

class Parent(Base):
    __tablename__ = 'parent'
    
    id = Column(Integer, primary_key=True)
    
    children = relationship("Child")
    
    def __repr__(self):
        return "<Parent(id='{}', children='{}')>".format(self.id, self.children)

class Child(Base):
    __tablename__ = 'child'
    
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))
    
    def __repr__(self):
        return "<Child(id='{}', parent_id='{}')>".format(self.id, self.parent_id)

 

5. 生成表

Base.metadata.drop_all(engine)
Base.metadata.create_all(engine) # 生產環境只需運行一次！！！！

 

6. 添加記錄

session.add_all([Parent(children=[Child() for j in range(5)]) for i in range(2)])
session.commit()

 

7. 查詢

# 懶加載 (lazyload)
# load everything, no eager loading.
for parent in session.query(Parent):
    print(parent.children)

# 聯合加載 (joinedload)
# load everything, joined eager loading.
for parent in session.query(Parent).options(joinedload("children")):
    parent.children

# 子查詢加載 (subqueryload)
# load everything, subquery eager loading.
for parent in session.query(Parent).options(subqueryload("children")):
    parent.children

 

8. 完整代碼

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.orm import joinedload, subqueryload


# 建立連接
engine = create_engine('sqlite:///:memory:', echo=True)


# 建立會話
session = Session(engine)


# 聲明基類
Base = declarative_base()


# 定義表（繼承基類）
class Parent(Base):
    __tablename__ = 'parent'
    
    id = Column(Integer, primary_key=True)
    
    children = relationship("Child")
    
    def __repr__(self):
        return "<Parent(id='{}', children='{}')>".format(self.id, self.children)

class Child(Base):
    __tablename__ = 'child'
    
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))
    
    def __repr__(self):
        return "<Child(id='{}', parent_id='{}')>".format(self.id, self.parent_id)
        

# 生成表
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine) # 生產環境只需運行一次！！！！


# 添加記錄
session.add_all([Parent(children=[Child() for j in range(5)]) for i in range(2)])
session.commit()


# 查詢
# lazyload
# load everything, no eager loading.
for parent in session.query(Parent):
    print(parent.children)

# joinedload
# load everything, joined eager loading.
for parent in session.query(Parent).options(joinedload("children")):
    parent.children

# subqueryload
# load everything, subquery eager loading.
for parent in session.query(Parent).options(subqueryload("children")):
    parent.children