sqlalchemy 现有数据库查询
Posted
技术标签:
【中文标题】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 不会喜欢它。
【讨论】:
以上是关于sqlalchemy 现有数据库查询的主要内容,如果未能解决你的问题,请参考以下文章
将 pandas 数据框行绑定到 sqlAlchemy 自定义查询