执行 sqlalchemy 存在查询
Executing a sqlalchemy exists query
我无法理解如何执行查询来检查匹配记录是否已存在于 sqlalchemy 中。我可以在网上找到的大多数示例似乎都引用了我没有的 "session" 和 "query" 对象。
这里有一个简短的完整程序来说明我的问题:
1. 使用 "person" table.
设置内存中的 sqlite 数据库
2.插入两条记录到人table.
3. 检查 table 中是否存在特定记录。这就是它吐的地方。
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
from sqlalchemy.sql.expression import exists
engine = create_engine('sqlite:///:memory:', echo=False)
metadata = MetaData()
person = Table('person', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(255), nullable=False))
metadata.create_all(engine)
conn = engine.connect()
s = person.insert()
conn.execute(s, name="Alice")
conn.execute(s, name="Bob")
print("I can see the names in the table:")
s = person.select()
result = conn.execute(s)
print(result.fetchall())
print('This query looks like it should check to see if a matching record exists:')
s = person.select().where(person.c.name == "Bob")
s = exists(s)
print(s)
print("But it doesn't run...")
result = conn.execute(s)
这个程序的输出是:
I can see the names in the table:
[(1, 'Alice'), (2, 'Bob')]
This query looks like it should check to see if a matching record exists:
EXISTS (SELECT person.id, person.name
FROM person
WHERE person.name = :name_1)
But it doesn't run...
Traceback (most recent call last):
File "/project_path/db_test/db_test_env/exists_example.py", line 30, in <module>
result = conn.execute(s)
File "/project_path/db_test/db_test_env/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 945, in execute
return meth(self, multiparams, params)
File "/project_path/db_test/db_test_env/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 265, in _execute_on_connection
raise exc.ObjectNotExecutableError(self)
sqlalchemy.exc.ObjectNotExecutableError: Not an executable object: <sqlalchemy.sql.selectable.Exists object at 0x105797438>
exists
用于 SQL 子查询。如果您有一个包含博客 post 的 table posts
和一个 author_id,映射回人,您可以使用如下查询来查找创建博客的人post:
select * from people where exists (select author_id from posts where author_id = people.id);
您不能将 exists 作为 SQL 查询中的最外层语句;它是在 SQL 布尔子句中使用的运算符。
因此,SQLAlchemy 不允许您执行该查询,因为它的格式不正确。
如果要查看某行是否存在,只需构造一个带where子句的select语句,然后查看查询的行数returns.
s.exists() 仅构建 exists 子句。要使代码正常工作,您需要做的就是为其生成 select。
s = exists(s).select()
这是您的完整示例:
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
from sqlalchemy.sql.expression import exists
engine = create_engine('sqlite:///:memory:', echo=False)
metadata = MetaData()
person = Table('person', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(255), nullable=False))
metadata.create_all(engine)
conn = engine.connect()
s = person.insert()
conn.execute(s, name="Alice")
conn.execute(s, name="Bob")
print("I can see the names in the table:")
s = person.select()
result = conn.execute(s)
print(result.fetchall())
print('This query looks like it should check to see if a matching record exists:')
s = person.select().where(person.c.name == "Bob")
s = exists(s).select()
print(s)
print("And it runs fine...")
result = conn.execute(s)
print(result.fetchall())
试试这个:
...
s = person.select().where(person.c.name == "Bob")
s = select(exists(s))
print(s)
...
除非有人提出更好的答案,否则这就是我想出的有效答案。让数据库计算匹配记录并将计数发送到 python 应用程序。
from sqlalchemy import select, func # more imports not in my example code above
s = select([func.count(1)]).select_from(person).where(person.c.name == "Bob")
print(s)
record_count = conn.execute(s).scalar()
print("Matching records: ", record_count)
示例输出:
SELECT count(:count_2) AS count_1
FROM person
WHERE person.name = :name_1
Matching records: 1
我无法理解如何执行查询来检查匹配记录是否已存在于 sqlalchemy 中。我可以在网上找到的大多数示例似乎都引用了我没有的 "session" 和 "query" 对象。
这里有一个简短的完整程序来说明我的问题:
1. 使用 "person" table.
设置内存中的 sqlite 数据库
2.插入两条记录到人table.
3. 检查 table 中是否存在特定记录。这就是它吐的地方。
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
from sqlalchemy.sql.expression import exists
engine = create_engine('sqlite:///:memory:', echo=False)
metadata = MetaData()
person = Table('person', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(255), nullable=False))
metadata.create_all(engine)
conn = engine.connect()
s = person.insert()
conn.execute(s, name="Alice")
conn.execute(s, name="Bob")
print("I can see the names in the table:")
s = person.select()
result = conn.execute(s)
print(result.fetchall())
print('This query looks like it should check to see if a matching record exists:')
s = person.select().where(person.c.name == "Bob")
s = exists(s)
print(s)
print("But it doesn't run...")
result = conn.execute(s)
这个程序的输出是:
I can see the names in the table:
[(1, 'Alice'), (2, 'Bob')]
This query looks like it should check to see if a matching record exists:
EXISTS (SELECT person.id, person.name
FROM person
WHERE person.name = :name_1)
But it doesn't run...
Traceback (most recent call last):
File "/project_path/db_test/db_test_env/exists_example.py", line 30, in <module>
result = conn.execute(s)
File "/project_path/db_test/db_test_env/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 945, in execute
return meth(self, multiparams, params)
File "/project_path/db_test/db_test_env/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 265, in _execute_on_connection
raise exc.ObjectNotExecutableError(self)
sqlalchemy.exc.ObjectNotExecutableError: Not an executable object: <sqlalchemy.sql.selectable.Exists object at 0x105797438>
exists
用于 SQL 子查询。如果您有一个包含博客 post 的 table posts
和一个 author_id,映射回人,您可以使用如下查询来查找创建博客的人post:
select * from people where exists (select author_id from posts where author_id = people.id);
您不能将 exists 作为 SQL 查询中的最外层语句;它是在 SQL 布尔子句中使用的运算符。 因此,SQLAlchemy 不允许您执行该查询,因为它的格式不正确。 如果要查看某行是否存在,只需构造一个带where子句的select语句,然后查看查询的行数returns.
s.exists() 仅构建 exists 子句。要使代码正常工作,您需要做的就是为其生成 select。
s = exists(s).select()
这是您的完整示例:
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
from sqlalchemy.sql.expression import exists
engine = create_engine('sqlite:///:memory:', echo=False)
metadata = MetaData()
person = Table('person', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(255), nullable=False))
metadata.create_all(engine)
conn = engine.connect()
s = person.insert()
conn.execute(s, name="Alice")
conn.execute(s, name="Bob")
print("I can see the names in the table:")
s = person.select()
result = conn.execute(s)
print(result.fetchall())
print('This query looks like it should check to see if a matching record exists:')
s = person.select().where(person.c.name == "Bob")
s = exists(s).select()
print(s)
print("And it runs fine...")
result = conn.execute(s)
print(result.fetchall())
试试这个:
...
s = person.select().where(person.c.name == "Bob")
s = select(exists(s))
print(s)
...
除非有人提出更好的答案,否则这就是我想出的有效答案。让数据库计算匹配记录并将计数发送到 python 应用程序。
from sqlalchemy import select, func # more imports not in my example code above
s = select([func.count(1)]).select_from(person).where(person.c.name == "Bob")
print(s)
record_count = conn.execute(s).scalar()
print("Matching records: ", record_count)
示例输出:
SELECT count(:count_2) AS count_1
FROM person
WHERE person.name = :name_1
Matching records: 1