Sqlalchemy query execution exists
I am having trouble figuring out how to execute a validation query and see if the corresponding entry exists in sqlalchemy. Most of the examples I can find on the Internet seem to refer to the "session" and "request" objects, which I don't have.
Here is a short complete program that illustrates my problem:
1. installs the "person" table in the sqlite db in memory.
2. Inserts two records into the faces table. 3. check if a particular record exists in the table. That's where barfs is.
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)
The output of this program:
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>
source to share
s.exists () only builds an existence condition. All you have to do to make your code work is generate a selection for it.
s = exists(s).select()
Here's your complete example:
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())
source to share
exists
used in SQL subqueries. If you have a table posts
containing a blog post with author_id, back to people, you can use the following query to find the people who made the blog post:
select * from people where exists (select author_id from posts where author_id = people.id);
You cannot have an externally existing statement in an SQL query; it is a statement for use in boolean SQL statements. Thus, SQLAlchemy does not allow this query to be executed because it is not valid. If you want to see if a row exists, just create a select statement with a where clause and see how many rows the query returns.
source to share
Unless someone suggests a better answer, here is what I came up with works. Having a DB counts the corresponding records and only sends the invoice to the python app.
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)
Output example:
SELECT count(:count_2) AS count_1
FROM person
WHERE person.name = :name_1
Matching records: 1
source to share