python调试技巧：数据库调试，转换sqlalchemy中的sql语句在Python代码中对数据查询使⽤sqlalchemy
但是在对代码进⾏检查调试的时候，需要对将查询语句转换成纯sql语句，放⼊数据库中查询调试
查询数据为query_str = session.query(User).filter(User.id == 1)时
print query_str
结果为
SELECT users.id AS users_id, users.name AS users_name, ail AS users_email
FROM users
WHERE users.id = :id_1
打印的结果⽆法将filter(User.id == 1)中的1进⾏转换，需要⼿⼯修改
现添加函数literalquery可将查询语句直接转换成sql语句
代码如下：
# -*- coding: utf-8 -*-
ine.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.ding)
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,
}
# 打印执⾏的sql语句，print(literalquery(query))
def literalquery(statement):
def literalquery(statement):
"""NOTE: This is entirely insecure. DO NOT execute the resulting strings."""
if isinstance(statement, Query):
statement = statement.statement
return statementpile(
dialect=LiteralDialect(),
compile_kwargs={'literal_binds': True},
).string
if __name__ == "__main__":
from flask import Flask
from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
import Session
declarative import declarative_base
app =  Flask(__name__)
Base = declarative_base()
# 定义ORM
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(50), unique=True)
email = Column(String(120), unique=True)
def __init__(self, name=None, email=None):
self.name = name
def __repr__(self):
return '<User %r>' % (self.name)
engine = create_engine('sqlite:///./sqlalchemy1.db')
global session
session = Session(engine)
python代码转换
admin = User('admin', 'admin@example')
session.add(admin)
guestes = [User('guest1', 'guest1@example'),
User('guest2', 'guest2@example'),
User('guest3', 'guest3@example'),
User('guest4', 'guest4@example')]
session.add_all(guestes)
sessionmit()
query_str = session.query(User).filter(User.id == 1)
print query_str
# 打印结果
# SELECT users.id AS users_id, users.name AS users_name, ail AS users_email
# FROM users
# WHERE users.id = :id_1
print literalquery(query_str) # 打印结果# SELECT users.id, users.name, ail # FROM users # WHERE users.id = 1

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。