How do I get a raw, compiled SQL query from a SQLAlchemy expression?
This blogpost by Nicolas Cadou provides an updated answer.
Quoting from the blog post, this is suggested and worked for me:
from sqlalchemy.dialects import postgresql
print str(q.statement.compile(dialect=postgresql.dialect()))
Where q is defined as:
q = DBSession.query(model.Name).distinct(model.Name.value) \
.order_by(model.Name.value)
Or just any kind of session.query()
.
SQLAlchemy: print the actual query
This works in python 2 and 3 and is a bit cleaner than before, but requires SA>=1.0.
from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.sql.sqltypes import String, DateTime, NullType
# python2/3 compatible.
PY3 = str is not bytes
text = str if PY3 else unicode
int_type = int if PY3 else (int, long)
str_type = str if PY3 else (str, unicode)
class StringLiteral(String):
"""Teach SA how to literalize various things."""
def literal_processor(self, dialect):
super_processor = super(StringLiteral, self).literal_processor(dialect)
def process(value):
if isinstance(value, int_type):
return text(value)
if not isinstance(value, str_type):
value = text(value)
result = super_processor(value)
if isinstance(result, bytes):
result = result.decode(dialect.encoding)
return result
return process
class LiteralDialect(DefaultDialect):
colspecs = {
# prevent various encoding explosions
String: StringLiteral,
# teach SA about how to literalize a datetime
DateTime: StringLiteral,
# don't format py2 long integers to NULL
NullType: StringLiteral,
}
def literalquery(statement):
"""NOTE: This is entirely insecure. DO NOT execute the resulting strings."""
import sqlalchemy.orm
if isinstance(statement, sqlalchemy.orm.Query):
statement = statement.statement
return statement.compile(
dialect=LiteralDialect(),
compile_kwargs={'literal_binds': True},
).string
Demo:
# coding: UTF-8
from datetime import datetime
from decimal import Decimal
from literalquery import literalquery
def test():
from sqlalchemy.sql import table, column, select
mytable = table('mytable', column('mycol'))
values = (
5,
u'snowman: ☃',
b'UTF-8 snowman: \xe2\x98\x83',
datetime.now(),
Decimal('3.14159'),
10 ** 20, # a long integer
)
statement = select([mytable]).where(mytable.c.mycol.in_(values)).limit(1)
print(literalquery(statement))
if __name__ == '__main__':
test()
Gives this output: (tested in python 2.7 and 3.4)
SELECT mytable.mycol
FROM mytable
WHERE mytable.mycol IN (5, 'snowman: ☃', 'UTF-8 snowman: ☃',
'2015-06-24 18:09:29.042517', 3.14159, 100000000000000000000)
LIMIT 1
executing a raw sql query from sqlalchemy on postgresql
As explained is the SQLAlchemy documentation, the .execute()
method returns only a proxy on which you'll have to iterate (or apply any aggregation method) to view the actual result of the query. Apparently, in your case, what you want is the .fetchall()
method.
If you try something like this:
from sqlalchemy import create_engine
engine = create_engine('/path/to/your/db...')
connection = engine.connect()
my_query = 'SELECT * FROM my_table'
results = connection.execute(my_query).fetchall()
the results
variable will be a list
of all the items that the query fetches.
Hope this helps!
Compile query from raw string (without using .text(...)) using Sqlalchemy connection and Postgres
Ok I think I got it.
The combination of SqlAlchemy's raw_connection
+ Psycopg's mogrify
seems to be the answer.
conn = sqlalch_engine.raw_connection()
try:
cursor = conn.cursor()
s_str = cursor.mogrify(statement_str, {'user_id': self.user_id})
s_str = s_str.decode("utf-8") # mogrify returns bytes
# Some cleanup for niceness:
s_str = s_str.replace('\n', ' ')
s_str = re.sub(r'\s{2,}', ' ', s_str)
finally:
conn.close()
I hope someone else finds this helpful
SQLAlchemy printing raw SQL from create()
from sqlalchemy.schema import CreateTable
print(CreateTable(table))
If you are using declarative syntax:
print(CreateTable(Model.__table__))
Update:
Since I have the accepted answer and there is important information in klenwell answer, I'll also add it here.
You can get the SQL for your specific database (MySQL, Postgresql, etc.) by compiling with your engine.
print(CreateTable(Model.__table__).compile(engine))
Update 2:
@jackotonye Added in the comments a way to do it without an engine.
print(CreateTable(Model.__table__).compile(dialect=postgresql.dialect()))
Retrieve query results as dict in SQLAlchemy
results is a tuple and I want it to be of type dict()
Updated answer for SQLAlchemy 1.4:
Version 1.4 has deprecated the old engine.execute()
pattern and changed the way .execute()
operates internally. .execute()
now returns a CursorResult object with a .mappings() method:
import sqlalchemy as sa
# …
with engine.begin() as conn:
qry = sa.text("SELECT FirstName, LastName FROM clients WHERE ID < 3")
resultset = conn.execute(qry)
results_as_dict = resultset.mappings().all()
pprint(results_as_dict)
"""
[{'FirstName': 'Gord', 'LastName': 'Thompson'},
{'FirstName': 'Bob', 'LastName': 'Loblaw'}]
"""
(Previous answer for SQLAlchemy 1.3)
SQLAlchemy already does this for you if you use engine.execute
instead of raw_connection()
. With engine.execute
, fetchone
will return a SQLAlchemy Row
object and fetchall
will return a list
of Row
objects. Row
objects can be accessed by key, just like a dict
:
sql = "SELECT FirstName, LastName FROM clients WHERE ID = 1"
result = engine.execute(sql).fetchone()
print(type(result)) # <class 'sqlalchemy.engine.result.Row'>
print(result['FirstName']) # Gord
If you need a true dict
object then you can just convert it:
my_dict = dict(result)
print(my_dict) # {'FirstName': 'Gord', 'LastName': 'Thompson'}
How to get the compiled raw sql from graphQL with SQLAlchemy?
Well, I just found the way, reference from https://stackoverflow.com/a/36141722/9041712
query = RecordType.get_query(info)
print(query.statement.compile(compile_kwargs={"literal_binds": True}))
Related Topics
How to Get Item's Position in a List
How to Hide the Console Window in a Pyqt App Running on Windows
Speed of Calculating Powers (In Python)
How Does a Python for Loop with Iterable Work
Call Int() Function on Every List Element
Failed to Catch Syntax Error Python
Pythonic Way to Check If a File Exists
Numpy - Create Matrix with Rows of Vector
Split List into Smaller Lists (Split in Half)
Why Sum on Lists Is (Sometimes) Faster Than Itertools.Chain
Strange Behavior of Lists in Python
How to Dump a Dict to a JSON File
Get _Name_ of Calling Function's Module in Python
How to Set the Figure Title and Axes Labels Font Size in Matplotlib