sqlalchemy 中使用 sql 查詢資料庫

• 背景
  有時候 ，我們希望通過原生 sql 來查詢資料庫，這個時候 應該怎麼辦呢？

sqlalchemy 已經給我們提供了這樣的介面， 通過 text 就可以輕鬆實現了。

• 例1

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
@Time    : 2018/12/8 10:03
@File    : query_data5.py
@Author  : [email protected]
"""

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine, text, Integer, Float

from secure import XINYONGFEI_BI_URL


def get_session():
    engine = create_engine(XINYONGFEI_BI_URL, pool_size=10, pool_recycle=7200,
                           pool_pre_ping=True, encoding='utf-8')

    session_factory = sessionmaker(bind=engine)
    session = session_factory()

    return session


if __name__ == '__main__':
    session = get_session()

    sql = """\
       select  `id`, `name`, `mobile`
    from  `TUser`  limit 5 ;
       """
    print(f'sql:{sql}')
    try:
        resultproxy = session.execute(
            text(sql)
        )
    except Exception as e:
        print(e)
        results = []
    else:
        results = resultproxy.fetchall()

    print(results)


 

sql:       select  `id`, `name`, `mobile`
    from  `TUser`  limit 5 ;
       
[(1.0, 'frank0', '123492130'), (2.0, 'frank1', '123492131'), (3.0, 'frank2', '123492132'), (4.0, 'frank3', '123492133'), (5.0, 'frank4', '123492134')]

這裡沒有引數 直接用 text 包裝一下， 就可以執行了。

• 例子2

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
@Time    : 2018/11/26 11:15
@File    : test_use_sql.py
@Author  : 
 
[email protected]

site-packages/sqlalchemy/orm/session.py

"""

from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine, text

from secure import XINYONGFEI_BI_URL


def to_dict(data):
    my_column = ["id", "order_number", "loan_set_id", "account_id", "amount", "success_time", "fail_time", "memo",
                 "order_memo", "order_status", "status", "created_time", "updated_time"]
    return dict(zip(my_column, data))


def get_session():
    engine = create_engine(XINYONGFEI_BI_URL, pool_size=10, pool_recycle=7200,
                           pool_pre_ping=True, encoding='utf-8')

    session_factory = sessionmaker(bind=engine)
    session = session_factory()

    return session


def get_table(table_name):
    """
    通過表名，對映資料庫裡面的表
    :param table_name:  str
    :return:
    """
    Base = automap_base()

    engine = create_engine(XINYONGFEI_BI_URL, pool_size=10, pool_recycle=7200,
                           pool_pre_ping=True, encoding='utf-8')

    Base.prepare(engine, reflect=True)
    classes = Base.classes

    try:
        table = getattr(classes, table_name)
        return table
    except AttributeError:
        raise AttributeError(f'not exist table_name:{table_name}')


if __name__ == '__main__':
    pass
    session = get_session()

    sql = """
    select  `id`, `order_number`, `loan_set_id`, `account_id`, `amount`, `success_time`, `fail_time`, `memo`, 
    `order_memo`, `order_status`, `status`, `created_time`, `updated_time`  from BidOrder
    where  account_id  = :account_id and amount = :amount ;
    """

    # print(f'sql:{sql}')
    try:
        resultproxy = session.execute(
            text(sql), {"account_id": 100000, "amount": 500}
        )
    except Exception as e:
        print(e)
        results = []
    else:
        results = resultproxy.fetchall()

    for result in results:
        print(to_dict(result))




"""
傳入引數 :param 

這裡傳入兩個引數
:account_id ,
:amount  

sql 要通過 text 包裝一下，  這樣就可以 通過 session 物件  的 execute 方法，向資料庫裡面查詢，寫入資料了。
注意引數是一個字典， key 就是 sql 中定義的引數


session.execute(
            text(sql), {"account_id": 100000, "amount": 500}
        )
"""



 

這裡執行sql 如下：

SELECT `id AS `, `order_number AS `, `loan_set_id AS `, `account_id AS `, `amount AS `
	, `success_time AS `, `fail_time AS `, `memo AS `, `order_memo AS `, `order_status AS `
	, `status AS `, `created_time AS `, `updated_time AS `
FROM BidOrder
WHERE account_id = 100000
	AND amount = 500;
	

總結

本文主要講了如何使用sqlalchemy 執行原生的sql, 主要是通過text 這個 物件來實現的。 當然還可以直接繫結引數，定製column 更多可以參考官網給的例子，例子還是非常豐富的.

參考文件

text 的使用 https://docs.sqlalchemy.org/en/latest/orm/tutorial.html#using-textual-sql

sql Expressions https://docs.sqlalchemy.org/en/latest/core/sqlelement.html#sqlalchemy.sql.expression.text

session 的基本操作，查詢程式碼，已經寫得比較清楚了。
site-packages/sqlalchemy/orm/session.py