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()

注釋：模型類自己定義，繼承生成的基類