sqlalchemy 執行原生sql語句

from contextlib import contextmanager
from sqlalchemy import create_engine, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship, scoped_session
from sqlalchemy.orm import sessionmaker


def create_session(engine):
    """封裝Session建立過程   此函數只在應用初始化時調用一次便可"""

    # 建立Session工廠     SessionFactory()就能夠建立出session對象, 可是該對象沒有實現線程隔離
    SessionFactory = sessionmaker(bind=engine)

    # 生成線程隔離的session對象(每一個線程取本身的session)
    session = scoped_session(SessionFactory)
    # ps: 其實返回的session是scoped_session類型對象, 再經過session()纔會建立Session對象, 不過scoped_session實現了代理功能, 進行數據操做時, 會自動建立/獲取實現了線程隔離的Session對象並執行操做

    return session


# 建立模型基類
Base = declarative_base()

# 建立數據庫引擎
# 細節1  sqlalchemy默認實現了線程池功能, 而且能夠自動重連(pool_size 線程池中的鏈接數, max_overflow 超出線程池的額外鏈接數)
# 細節2  若是數據庫使用utf-8支持中文, 則設置鏈接地址時須要標明  ?charset=utf8   不然報錯
engine = create_engine('mysql://root:mysql@127.0.0.1:3306/test27?charset=utf8', echo=False, pool_size=5, max_overflow=10)

# 建立會話
session = create_session(engine)


class User(Base):
    __tablename__ = 't_user'
    id = Column(Integer, primary_key=True)
    name = Column(String(20), unique=True)
    addresses = relationship('Address')


class Address(Base):
    __tablename__ = 't_address'
    id = Column(Integer, primary_key=True)
    detail = Column(String(20))
    user_id = Column(Integer, ForeignKey('t_user.id'))


@contextmanager  # 裝飾器形式的上下文管理器
def session_scope():
    """使用上下文管理器對session和事務操做進行封裝"""
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.remove()  # session工做完成, 銷燬session, 釋放內存


def index():
    """模擬視圖函數"""

    with session_scope() as session:  # 獲取session對象, 代碼塊執行完會自動提交 並 銷燬session
        """增長數據"""
        user1 = User(name='zs')
        session.add(user1)
        session.flush()

        adr1 = Address(detail="中關村3號", user_id=user1.id)
        adr2 = Address(detail="華強北5號", user_id=user1.id)
        session.add_all([adr1, adr2])

        """查詢數據"""
        # ret = session.query(User, Address).join(Address, User.id == Address.user_id).filter(User.name == 'zs').all()
        # for user, adr in ret:
        #     print(user.name, adr.detail)


        """執行原生SQL"""
        # data = session.execute('select * from t_user')
        # print(type(data))
        # row = data.fetchone()  # 取第一條
        # print(row.id)  # 取主鍵
        # print(row.name)  # 取字段

        # rows = data.fetchall()  # 取全部數據
        # for row in rows:
        #     print(row.id, row.name)

        # print(data.rowcount)  # 取條數


if __name__ == '__main__':
    # 刪除全部表
    Base.metadata.drop_all(engine)
    # 建立全部表
    Base.metadata.create_all(engine)

    index()

註釋：模型類本身定義，繼承生成的基類