sqlalchemy现有数据库查询

Question

我正在使用 SQLAlchemy 作为 python 项目的 ORM。我创建了几个 models/schema 并且工作正常。现在我需要查询现有的 MySQL 数据库，没有 insert/update 只是 select 语句。

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

例如，如果现有数据库有 table name User 那么我想使用 dbsession 查询它（只有 select 操作，可能有连接）

Answer 1

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

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

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

另一种方法是像往常一样简单地定义模型，注意定义模型以匹配数据库 tables，这并不难。这种方法的好处是您可以仅将数据库 table 的一个子集映射到您的模型，甚至可以将 table 列的一个子集映射到模型的字段。假设您在数据库中有 10 个 table，但只对 users table 感兴趣，而您只需要 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 权限。 Alternatively/in 此外，您还可以尝试在您的应用程序代码中回滚事务。

Answer 2

您可以使用 automap extension:

访问现有的 table

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()

Answer 3

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

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)

Answer 4

其他答案没有提到如果您有一个没有主键的 table 该怎么办，所以我想我会解决这个问题。假设一个名为 Customers 的 table 包含 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)

Answer 5

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

的 table

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()

Warning: Your table should have a Primary Key defined. Otherwise, Sqlalchemy won't like it.

sqlalchemy现有数据库查询

sqlalchemy existing database query

python

pylons

sqlalchemy

pyramid