本文主要Sqlalchemy的ORM components進行介紹
參考：SQLAlchemy 1.4 Documentation（中文版）

基本架構(gòu)

SQLAlchemy由幾個組件組成，包括SQL expression language和ORM。
為了實現(xiàn)這些組件，SQLAlchemy還提供了一個Engine類和MetaData類。

• Engine--管理SQLAlchemy連接池和獨立于數(shù)據(jù)庫的SQL方言層。
• MetaData - 用于收集和組織有關(guān)您的表布局（數(shù)據(jù)庫模式）的信息。
• SQL expression language--提供了一個API來對你的表執(zhí)行查詢和更新。
  所有這些都來自Python，而且都是以一種獨立于數(shù)據(jù)庫的方式。（本文不作介紹）
• ORM - 提供了一種方便的方式來為你的Python對象添加數(shù)據(jù)庫持久性，而不需要再添加其他的功能。

ORM創(chuàng)建表格

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base


# 建立一個引擎對象，如果數(shù)據(jù)庫不存在，將創(chuàng)建數(shù)據(jù)庫，下面以sqlite為例
engine = create_engine("sqlite:///test.db", echo=True)
# 創(chuàng)造基類的實例，通常只會有一個實例，建議做單例模式
Base = declarative_base(bind=engine)


class Customers(Base):
    """在Base的基礎上定義映射類

    Args:
        Base : 基類
    """    
    __tablename__ = 'customers'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    address = Column(String)
    email = Column(String)


def create_table():
    """創(chuàng)建還沒有被創(chuàng)建的表
    """    
    Base.metadata.create_all() 

獲取交互句柄

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("sqlite:///test.db", echo=True)

def get_session():
    """獲取與數(shù)據(jù)庫交互的句柄

    Returns:
        與數(shù)據(jù)庫交互的句柄
    """    
    Session = sessionmaker(bind=engine)
    session = Session()
    return session

數(shù)據(jù)庫操作

以新增customer為例

def add_customer(**kwargs):
    c = Customers(**kwargs)
    session = get_seesion()
    session.add(c)
    session.commit()
    session.close()
    return c

add_customer(name='xiaofan', address='xiaozhufeng', email='xiaofan@163.com')

其它常用操作

• add_all()
• query()

session.query(Customers).all()  # return list

session.query(Customers).first()  # return scalar
 # 超過一個報錯“MultipleResultsFound”， 少于一個報錯“NoResultFound”
session.query(Customers).filter(Customers.id == 1).one()   # retrun scalar

• filter()

# return list
session.query(Customers).filter(Customers.name=='yasin')  # ==
session.query(Customers).filter(Customers.id>2)  # >
session.query(Customers).filter(Customers.name.like('Ra%'))  # like
session.query(Customers).filter(Customers.id.in_([1,3]))  # in
session.query(Customers).filter(Customers.id>2,
Customers.name.like('Ra%'))  # and
session.query(Customers).filter(and_(Customers.id>2,
Customers.name.like('Ra%')))  # and
session.query(Customers).filter(or_(Customers.id>2,
Customers.name.like('Ra%')))  # or

• scalar()

session.query(Customers).filter(Customers.id==3).scalar()  # 僅當len(list)=1時有效

• update()

session.query(Customers).filter(Customers.id!=2).update({Customers.name:"Mr."+C
ustomers.name}, synchronize_session=False)

Textual SQL

• filter中使用SQL查詢語句

from sqlalchemy import text

session.query(Customers).filter(text("id<3"))
session.query(Customers).filter(text("id=:value")).params(value=1).one()

• 純SQL語句查詢

session.query(Customers).from_statement(text("SELECT * FROM customers")).all()

stmt = text("SELECT name, id, name, address, email FROM customers")
# name, id 對應鏈接到了Customers.id和Customers.name，互換了
stmt=stmt.columns(Customers.id, Customers.name) 
session.query(Customers.id, Customers.name).from_statement(stmt).all()