sqlalchemy 系列教程四使用原生sql 查询数据库

编程知识更新时间:2023-04-05 11:09:04

sqlalchemy 中使用 sql 查询数据库

背景
有时候，我们希望通过原生 sql 来查询数据库，这个时候应该怎么办呢？

sqlalchemy 已经给我们提供了这样的接口，通过 text 就可以轻松实现了。

例1

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
@Time    : 2018/12/8 10:03
@File    : query_data5.py
@Author  : frank.chang@shoufuyou
"""

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  : frank.chang@shoufuyou

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;