SQLAlchemy ORM 转换为 pandas DataFrame-IT科技

摘要：问题描述：有没有将 SQLAlchemy 转换<Query object>为 pandas DataFrame 的解决方案？Pandas 有能力使用，pandas.read_sql但这需要使用原始 SQL。我有两个理由想要避免它：我已经使用 ORM 做了所有事情（这本身就是个很好的理由），我使用 ...

问题描述：

有没有将 SQLAlchemy 转换<Query object>为 pandas DataFrame 的解决方案？

Pandas 有能力使用，pandas.read_sql但这需要使用原始 SQL。我有两个理由想要避免它：

我已经使用 ORM 做了所有事情（这本身就是个很好的理由），
我使用 python 列表作为查询的一部分，例如：

db.session.query(Item).filter(Item.symbol.in_(add_symbols)其中Item是我的模型类，add_symbols是一个列表）。这相当于 SQL SELECT ... from ... WHERE ... IN。

有什么可能吗？

解决方案 1：

以下应该在大多数情况下有效：

df = pd.read_sql(query.statement, query.session.bind)

pandas.read_sql有关参数的更多信息，请参阅文档。

解决方案 2：

为了让新手 Pandas 程序员更清楚地了解这一点，这里有一个具体的例子，

pd.read_sql(session.query(Complaint).filter(Complaint.id == 2).statement,session.bind)

这里我们从投诉表（sqlalchemy 模型是投诉）中选择一个投诉，id = 2

解决方案 3：

为了完整性：作为 Pandas 函数的替代read_sql_query()，您还可以使用 Pandas-DataFrame 函数from_records()来转换structured or record ndarray to DataFrame。

如果您已经在 SQLAlchemy 中执行了查询并且已经有了结果，那么这将非常方便：

import pandas as pd 
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker


SQLALCHEMY_DATABASE_URI = 'postgresql://postgres:postgres@localhost:5432/my_database'
engine = create_engine(SQLALCHEMY_DATABASE_URI, pool_pre_ping=True, echo=False)
db = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))
Base = declarative_base(bind=engine)


class Currency(Base):
    """The `Currency`-table"""
    __tablename__ = "currency"
    __table_args__ = {"schema": "data"}

    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(String(64), nullable=False)


# Defining the SQLAlchemy-query
currency_query = db.query(Currency).with_entities(Currency.id, Currency.name)

# Getting all the entries via SQLAlchemy
currencies = currency_query.all()

# We provide also the (alternate) column names and set the index here,
# renaming the column `id` to `currency__id`
df_from_records = pd.DataFrame.from_records(currencies
    , index='currency__id'
    , columns=['currency__id', 'name'])
print(df_from_records.head(5))

# Or getting the entries via Pandas instead of SQLAlchemy using the
# aforementioned function `read_sql_query()`. We can set the index-columns here as well
df_from_query = pd.read_sql_query(currency_query.statement, db.bind, index_col='id')
# Renaming the index-column(s) from `id` to `currency__id` needs another statement
df_from_query.index.rename(name='currency__id', inplace=True)
print(df_from_query.head(5))

解决方案 4：

所选的解决方案对我不起作用，因为我一直收到错误

AttributeError：'AnnotatedSelect' 对象没有属性 'lower'

我发现以下方法有效：

df = pd.read_sql_query(query.statement, engine)

解决方案 5：

如果您想使用参数和方言特定参数编译查询，请使用以下命令：

c = query.statement.compile(query.session.bind)
df = pandas.read_sql(c.string, query.session.bind, params=c.params)

解决方案 6：

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine('postgresql://postgres:postgres@localhost:5432/DB', echo=False)
Base = declarative_base(bind=engine)
Session = sessionmaker(bind=engine)
session = Session()

conn = session.bind

class DailyTrendsTable(Base):

    __tablename__ = 'trends'
    __table_args__ = ({"schema": 'mf_analysis'})

    company_code = Column(DOUBLE_PRECISION, primary_key=True)
    rt_bullish_trending = Column(Integer)
    rt_bearish_trending = Column(Integer)
    rt_bullish_non_trending = Column(Integer)
    rt_bearish_non_trending = Column(Integer)
    gen_date = Column(Date, primary_key=True)

df_query = select([DailyTrendsTable])

df_data = pd.read_sql(rt_daily_query, con = conn)

解决方案 7：

Result.keys()使用该方法获取列名的简单示例。

import sqlalchemy as sa
import pandas as pd

engine = sa.create_engine(...)
with engine.connect() as conn:
   result = conn.execute("SELECT * FROM foo;")
   df = pd.DataFrame(result.all(), columns=result.keys())

https://docs.sqlalchemy.org/en/20/core/connections.html#sqlalchemy.engine.Result.keys

解决方案 8：

使用2.0 SQLalchemy语法（在 1.4 中也可以通过标志使用future=True）看起来pd.read_sql尚未实现并且会引发：

NotImplementedError: This method is not implemented for SQLAlchemy 2.0.

这是一个悬而未决的问题，直到 pandas 2.0 才会解决，您可以在这里和这里找到有关此问题的一些信息。

我没有找到任何令人满意的解决方法，但有些人似乎正在使用两种引擎配置，一种带有标志 future False：

engine2 = create_engine(URL_string, echo=False, future=False)

如果您查询字符串，这个解决方案是可以的，但是使用 ORM，我能做的最好的就是自定义函数尚未优化，但它有效：

Conditions = session.query(ExampleTable)
def df_from_sql(query):
    return pd.DataFrame([i.__dict__ for i in query]).drop(columns='_sa_instance_state')
df = df_from_sql(ExampleTable)

无论如何，这个解决方案都是临时的，直到 pd.read_sql 实现了新的语法。

解决方案 9：

使用类似@van的答案进行补充read_sql，当我的查询涉及连接时，sqlalchemy似乎隐式地从连接表中添加别名列，例如id_1，id_2，以防连接表和主表都具有id列。使用.all()会在返回结果之前删除这些隐式列，但read_sql会包含这些列。

对我来说，这种情况的解决方案是明确我的选择。所以我替换了

query = session.query(model)

和

query = session.query(model.col_1, model.col_2)

或选择全部

query = session.query(*model.__table__.columns.values())

然后

df = pd.read_sql(query.statement, query.session.bind)

解决方案 10：

当你使用 ORM 时它就这么简单：

pd.DataFrame([r._asdict() for r in query.all()])

pd.read_sql当您不想将 SQL 和会话暴露给业务逻辑代码时，这是一个很好的替代方案。

在这里找到：https ://stackoverflow.com/a/52208023/1635525

解决方案 11：

此答案提供了一个可重现的示例，使用 SQL Alchemyselect语句并返回 pandas 数据框。它基于内存中的 SQLite 数据库，因此任何人都可以重现它而无需安装数据库引擎。

import pandas
from sqlalchemy import create_engine
from sqlalchemy import MetaData, Table, Column, Text
from sqlalchemy.orm import Session

定义表元数据并创建表

engine = create_engine('sqlite://')
meta = MetaData()
meta.bind = engine
user_table = Table('user', meta,
                   Column("name", Text),
                   Column("full_name", Text))
user_table.create()

user在表中插入一些数据

stmt = user_table.insert().values(name='Bob', full_name='Sponge Bob')
with Session(engine) as session:
    result = session.execute(stmt)
    session.commit()

将select语句的结果读入 pandas 数据框

# Select data into a pandas data frame
stmt = user_table.select().where(user_table.c.name == 'Bob')
df = pandas.read_sql_query(stmt, engine)
df
Out:
  name   full_name
0  Bob  Sponge Bob

解决方案 12：

如果使用 SQL 查询

def generate_df_from_sqlquery(query):
   from pandas import DataFrame
   query = db.session.execute(query)
   df = DataFrame(query.fetchall())
   if len(df) > 0:
      df.columns = query.keys()
   else:
      columns = query.keys()
      df = pd.DataFrame(columns=columns)
return df

profile_df = generate_df_from_sqlquery(profile_query)

解决方案 13：

如果您已经在 sqalchemy 过滤后收到了数据，则可以采用另一种方法来执行此操作，如下所示
按照此答案操作，然后使用此函数：

from collections import defaultdict
from sqlalchemy import inspect
import pandas as pd
def query_to_dict(rest):
    result = defaultdict(list)
    for obj in rest:
        instance = inspect(obj)
        for key, x in instance.attrs.items():
            result[key].append(x.value)
    return result
# Getting all the entries via SQLAlchemy
currencies = currency_query.all()
df = pd.DataFrame(query_to_dict(rset))

解决方案 14：

对我来说，这在 2023 年对我有用：首先创建一个引擎和会话，将查询字符串转换为 SQLalchemy 文本

from sqlalchemy import text

    engine = create_engine(path, echo=True)

    Session = sessionmaker(bind=engine)
    session = Session()

    # generates suitble text for SQLalchemy
    sql = text(query_string)

    session.execute(sql)

    df = pd.read_sql_query(sql=sql, con=engine.connect())

解决方案 15：

import pandas as pd

users = Users.query.all()

df = pd.DataFrame(users)

print(df)