SQLAlchemy 2.0:
with engine.connect() as connection:
result = connection.execute(text('SELECT * FROM your_table'))
# do something with the result..
SQLAlchemy 1.x:
from sqlalchemy import text
sql = text('select name from penguins')
result = db.engine.execute(sql)
names = [row[0] for row in result]
print names
Note that db.engine.execute() is "connectionless", which is deprecated in SQLAlchemy 2.0.
SQLAlchemy 2.0:
with engine.connect() as connection:
result = connection.execute(text('SELECT * FROM your_table'))
# do something with the result..
SQLAlchemy 1.x:
from sqlalchemy import text
sql = text('select name from penguins')
result = db.engine.execute(sql)
names = [row[0] for row in result]
print names
Note that db.engine.execute() is "connectionless", which is deprecated in SQLAlchemy 2.0.
SQL Alchemy session objects have their own execute method:
result = db.session.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})
All your application queries should be going through a session object, whether they're raw SQL or not. This ensures that the queries are properly managed by a transaction, which allows multiple queries in the same request to be committed or rolled back as a single unit. Going outside the transaction using the engine or the connection puts you at much greater risk of subtle, possibly hard to detect bugs that can leave you with corrupted data. Each request should be associated with only one transaction, and using db.session will ensure this is the case for your application.
Also take note that execute is designed for parameterized queries. Use parameters, like :val in the example, for any inputs to the query to protect yourself from SQL injection attacks. You can provide the value for these parameters by passing a dict as the second argument, where each key is the name of the parameter as it appears in the query. The exact syntax of the parameter itself may be different depending on your database, but all of the major relational databases support them in some form.
Assuming it's a SELECT query, this will return an iterable of RowProxy objects.
You can access individual columns with a variety of techniques:
for r in result:
print(r[0]) # Access by positional index
print(r['my_column']) # Access by column name as a string
r_dict = dict(r.items()) # convert to dict keyed by column names
Personally, I prefer to convert the results into namedtuples:
from collections import namedtuple
Record = namedtuple('Record', result.keys())
records = [Record(*r) for r in result.fetchall()]
for r in records:
print(r.my_column)
print(r)
If you're not using the Flask-SQLAlchemy extension, you can still easily use a session:
import sqlalchemy
from sqlalchemy.orm import sessionmaker, scoped_session
engine = sqlalchemy.create_engine('my connection string')
Session = scoped_session(sessionmaker(bind=engine))
s = Session()
result = s.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})
The tutorial gives a pretty good example for this:
>>> from sqlalchemy.sql import text
>>> s = text(
... "SELECT users.fullname || ', ' || addresses.email_address AS title "
... "FROM users, addresses "
... "WHERE users.id = addresses.user_id "
... "AND users.name BETWEEN :x AND :y "
... "AND (addresses.email_address LIKE :e1 "
... "OR addresses.email_address LIKE :e2)")
SQL>>> conn.execute(s, {"x": "m", "y": "z", "e1": "%@aol.com", "e2": "%@msn.com"}).fetchall()
[(u'Wendy Williams, [email protected]',)]
First, take your SQL string and pass it to sqalchemy.sql.text(). This isn't necessary, but probably a good idea...
The advantages text() provides over a plain string are backend-neutral support for bind parameters, per-statement execution options, as well as bind parameter and result-column typing behavior, allowing SQLAlchemy type constructs to play a role when executing a statement that is specified literally.
Note that even if you didn't use text(), you should NEVER just use sql.format(...). This leads to greater risk of SQL injection attacks.
Next, you can specify the actual arguments using keyword parameters to the execute() function you've already been using.
Now, in your example, you have a function that wraps the execute functionality. So, if you want to use this for multiple queries, you'll need to make the parameters able to receive your arguments. You could do this pretty simple as a dictionary:
def _sql_to_data(sql, values):
...
conn.execute(sql, values)
values would be a dictionary.You could then use your function like this...
sql = 'SELECT ...'
data = { 'user_id' : 3 }
results = _sql_to_data(sql, data)
Using keywords as your parameters is just one way of specifying the arguments to the execute() function. You can read the documentation for that function for a few different ways.
This is the latest way to bind params for session execute
stmt = text("SELECT * FROM attendance WHERE user_id = :x")
stmt = stmt.bindparams(x="1")
res = session.execute(stmt).all()