Python的sqlalchemy使用原生sql查詢如何返回字典形式的數組？

首先使用一個變量接收你以上的查詢結果。這個查詢結果是一個list，在這個list中包含着一個或多個tuple，其實這並不是標准的Python tuple，而是一個特殊的類型"<class 'sqlalchemy.util._collections.result'>"，這是一個 AbstractKeyedTuple 對象，它擁有一個 keys() 方法。我們可以通過這個方法將查詢結果轉換為字典，需要傳到前端展示只需要將其裝換為json格式即可。
示例：data = [dict(zip(result.keys(), result)) for result in results]

1.獲取session

#!/usr/bin/env python
# -*- coding:utf-8 -*-
__author__ = "JentZhang"
from utils.config import AppConfig
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine, text, event

# 創建對象的基類:
Base = declarative_base()


def get_yun_session():
    """
    初始化yun數據庫連接
    :return:
    """
    # '數據庫類型+數據庫驅動名稱://用戶名:口令@機器地址:端口號/數據庫名'
    engine = create_engine(
        "mysql+pymysql://%s:%s@%s:%s/%s?charset=utf8" % (
            AppConfig.USER, AppConfig.PASSWORD, AppConfig.HOST, AppConfig.PORT, AppConfig.DATABASE_YUN),
        max_overflow=AppConfig.MAX_OVERFLOW,
        pool_size=500,  # 連接池大小
        pool_timeout=30,  # 池中沒有線程最多等待的時間，否則報錯
        pool_recycle=100  # 多久之后對線程池中的線程進行一次連接的回收（重置）
    )
    session_maker = sessionmaker(bind=engine)
    return session_maker()

2.使用原生sql查詢

#!/usr/bin/env python
# -*- coding:utf-8 -*-
__author__ = "JentZhang"

from utils.mysql_utils import get_yun_dw_session, get_yun_session
from model import fac_broker, fac_message
from sqlalchemy import text


def get_user_list_dal(*args, **kwargs):
    """
    獲取用戶列表
    :param args:
    :param kwargs:
    :return:
    """
    session = get_yun_session()
    sql_str = """
                SELECT
                    a.user_id,
                    a.city_id,
                    a.`name`,
                    a.mobile,
                    DATE_FORMAT(a.create_time, '%Y-%m-%d') AS create_date,
                    IFNULL(
                        DATE_FORMAT(
                            a.login_first_time,
                            '%Y-%m-%d'
                        ),
                        ''
                    ) AS login_first_date,
                    IFNULL(
                        DATE_FORMAT(c.last_login, '%Y-%m-%d'),
                        ''
                    ) AS last_login_date,
                    (
                        CASE
                        WHEN isnull(d.user_id) THEN
                            '自然下載'
                        ELSE
                            d.channel_name
                        END
                    ) AS channel_name
                FROM
                    card_broker a
                LEFT JOIN fang_webgetcity b ON a.city_id = b.city_id
                LEFT JOIN account_user c ON a.user_id = c.id
                LEFT JOIN View_Broker_Channel d ON a.user_id = d.new_user_id
                where a.create_time>'2020-04-01'
    """
    res_rows = session.execute(text(sql_str)).fetchall()
    session.close()
    result = [dict(zip(result.keys(), result)) for result in res_rows]
    print(result)


if __name__ == '__main__':
    get_user_list_dal()

幾點提醒：

1.代碼中設計到的數據庫連接的配置文件：

class AppConfig(object):
    # 數據庫相關（測試）
    HOST = '127.0.0.1'
    PORT = 3306
    USER = 'root'
    PASSWORD = 'admin'
    DATABASE_YUN = 'yun'
    ECHO = False

    MAX_OVERFLOW = 500

2.對於使用原生SQL查詢出來的結果是一個list，

首先，使用一個變量接收你以上的查詢結果。

其次，在這個list中包含着一個或多個tuple，其實這並不是標准的Python tuple，而是一個特殊的類型"<class 'sqlalchemy.util._collections.result'>"，

　　    這是一個 AbstractKeyedTuple 對象，它擁有一個 keys() 方法。

最后，我們可以通過這個方法將查詢結果轉換為字典，需要傳到前端展示只需要將其裝換為json格式即可。

示例：data = [dict(zip(result.keys(), result)) for result in results]