執行sqlalchemy存在查詢

Question

我無法理解如何執行查詢來檢查並查看sqlalchemy中是否存在匹配的記錄。我可以在網上找到的大多數例子似乎都引用了我沒有的「會話」和「查詢」對象。

下面是一個簡短的完整程序，說明我的問題：
1.設置帶有「人員」表的內存sqlite數據庫。
2.將兩條記錄插入到人員表中。
3.檢查表中是否存在特定的記錄。這是它的地方。

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> 

Answer 1

s.exists（）僅構建exists子句。所有你需要做的工作就是爲你的代碼生成一個選擇。

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

Answer 2

exists在SQL子查詢時使用。如果您有包含博客文章與AUTHOR_ID表posts，映射回的人，你可能會使用一個查詢類似下面找到誰做了一個博客帖子的人：

select * from people where exists (select author_id from posts where author_id = people.id); 

你不能有一個存在作爲SQL查詢中的最外層語句;它是一個在SQL布爾子句中使用的運算符。 因此，SQLAlchemy不會讓你執行該查詢，因爲它不是格式良好的。 如果您想查看是否存在行，只需使用where子句構造select語句並查看查詢返回的行數。

Answer 3

試試這個：

... 
s = person.select().where(person.c.name == "Bob") 
s = select(exists(s)) 
print(s) 
... 

Answer 4

除非有人提出一個更好的答案，這裏是我想出的作品。讓數據庫統計匹配記錄並將計數發送到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