sqlalchemy 现有数据库查询

Posted 2023-03-16

【中文标题】sqlalchemy 现有数据库查询

【英文标题】：sqlalchemy existing database query

【发布时间】：2017-02-18 17:06:05

【问题描述】：

我正在使用 SQLAlchemy 作为 Python 项目的 ORM。我创建了几个模型/模式，它工作正常。现在我需要查询一个现有的 mysql 数据库，没有插入/更新只是 select 语句。

如何围绕现有数据库的表创建包装器？我已经简要浏览了 sqlalchemy 文档和 SO，但找不到任何相关内容。都建议执行方法，我需要在其中编写原始 sql 查询，而我想以与 SA 模型相同的方式使用 SQLAlchemy 查询方法。

例如，如果现有 db 有表名 User 那么我想使用 dbsession 查询它（只有 select 操作，可能带有 join）

【问题讨论】：

如果您想使用映射类，您可以使用reflection 和automap。

【参考方案1】：

您似乎认为 SQLAlchemy 只能使用由 SQLAlchemy 创建的数据库结构（可能使用MetaData.create_all()）——这是不正确的。 SQLAlchemy 可以与预先存在的数据库完美配合，您只需要定义模型以匹配数据库表。正如 Ilja Everilä 建议的那样，一种方法是使用反射：

class MyClass(Base):
    __table__ = Table('mytable', Base.metadata,
                    autoload=True, autoload_with=some_engine)

（在我看来，这对于一次性脚本来说完全没问题，但如果数据库结构可能随时间发生变化，则可能会在“真实”应用程序中导致令人难以置信的令人沮丧的错误）

另一种方法是像往常一样简单地定义模型，注意定义模型以匹配数据库表，这并不难。这种方法的好处是您只能将数据库表的子集映射到模型，甚至只能将表列的子集映射到模型的字段。假设您在数据库中有 10 个表，但只对 users 表感兴趣，而您只需要 id、name 和 email 字段：

class User(Base):
    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String)
    email = sa.Column(sa.String)

（请注意我们不需要定义一些只需要发出正确 DDL 的细节，例如字符串字段的长度或 email 字段具有索引的事实）

除非您在代码中创建或修改模型，否则 SQLAlchemy 不会发出 INSERT/UPDATE 查询。如果您想确保您的查询是只读的，您可以在数据库中创建一个特殊用户并仅授予该用户 SELECT 权限。或者/此外，您还可以尝试在应用程序代码中回滚事务。

【讨论】：

谢谢，虽然我使用了带反射的自动映射，但你的建议看起来也不错。对于 SO 用户，如果您有时间，您还可以添加一个 Automap 示例。这将使这个答案完整

@Sergey 你能添加所有的导入吗？

@Sergey 是的，导入可能很重要（双关语）。

【参考方案2】：

创建一个启用自动加载的表来检查它。一些示例代码：

from sqlalchemy.sql import select
from sqlalchemy import create_engine, MetaData, Table

CONN_STR = '…'
engine = create_engine(CONN_STR, echo=True)
metadata = MetaData()
cookies = Table('cookies', metadata, autoload=True,
                           autoload_with=engine)
cols = cookies.c


with engine.connect() as conn:

    query = (
        select([cols.created_at, cols.name])
                .order_by(cols.created_at)
                .limit(1)
    )
    for row in conn.execute(query):
        print(row)

【讨论】：

autoload 已在 1.4 (docs.sqlalchemy.org/en/14/core/…) 中弃用。使用autoload_with。

【参考方案3】：

您可以使用automap extension 访问现有表：

from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session

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

Users = Base.classes.users
session = Session(engine)

res = session.query(Users).first()

【参考方案4】：

其他答案没有提到如果您的表没有主键该怎么办，所以我想我会解决这个问题。假设有一个名为 Customers 的表，其中包含 CustomerId、CustomerName、CustomerLocation 的列，您可以这样做；

from sqlalchemy.ext.automap import automap_base
from sqlalchemy import create_engine, MetaData, Column, String, Table
from sqlalchemy.orm import Session

Base = automap_base()

conn_str = '...'    
engine = create_engine(conn_str)
metadata = MetaData()
# you only need to define which column is the primary key. It can automap the rest of the columns.
customers = Table('Customers',metadata, Column('CustomerId', String, primary_key=true), autoload=True, autoload_with=engine)
Base.prepare()
Customers= Base.classes.Customers

session = Session(engine)

customer1 = session.query(Customers).first()
print(customer1.CustomerName)

【参考方案5】：

假设我们有一个名为 accounts 的 Postgresql 数据库。我们已经有一个名为users 的表。

import sqlalchemy as sa

psw = "verysecret"
db = "accounts"

# create an engine
pengine = sa.create_engine('postgresql+psycopg2://postgres:' + psw +'@localhost/' + db)

from sqlalchemy.ext.declarative import declarative_base
# define declarative base
Base = declarative_base()

# reflect current database engine to metadata
metadata = sa.MetaData(pengine)
metadata.reflect()

# build your User class on existing `users` table
class User(Base):
    __table__ = sa.Table("users", metadata)
    
# call the session maker factory
Session = sa.orm.sessionmaker(pengine)
session = Session()

# filter a record 
session.query(User).filter(User.id==1).first()

警告：您的表应定义主键。否则，Sqlalchemy 不会喜欢它。