Поскольку я упомянул об этом в комментарии, вот оболочка C ++ / CLI:
#include <windows.h>
namespace JDanielSmith
{
public ref class Utilities abstract sealed /* abstract sealed = static */
{
public:
CA_SUPPRESS_MESSAGE("Microsoft.Security", "CA2122:DoNotIndirectlyExposeMethodsWithLinkDemands")
static void SetSystemTime(System::DateTime dateTime) {
LARGE_INTEGER largeInteger;
largeInteger.QuadPart = dateTime.ToFileTimeUtc(); // "If your compiler has built-in support for 64-bit integers, use the QuadPart member to store the 64-bit integer."
FILETIME fileTime; // "...copy the LowPart and HighPart members [of LARGE_INTEGER] into the FILETIME structure."
fileTime.dwHighDateTime = largeInteger.HighPart;
fileTime.dwLowDateTime = largeInteger.LowPart;
SYSTEMTIME systemTime;
if (FileTimeToSystemTime(&fileTime, &systemTime))
{
if (::SetSystemTime(&systemTime))
return;
}
HRESULT hr = HRESULT_FROM_WIN32(GetLastError());
throw System::Runtime::InteropServices::Marshal::GetExceptionForHR(hr);
}
};
}
Клиентский код C # теперь очень прост:
JDanielSmith.Utilities.SetSystemTime(DateTime.Now);
Этот блог содержит обновленный ответ.
Цитата из сообщения в блоге, это предлагается и работает для меня.
>>> from sqlalchemy.dialects import postgresql
>>> print str(q.statement.compile(dialect=postgresql.dialect()))
Где q определяется как:
>>> q = DBSession.query(model.Name).distinct(model.Name.value) \
.order_by(model.Name.value)
Или просто любой вид session.query ().
Спасибо Николасу Каду за ответ! Надеюсь, это поможет другим, кто ищет здесь.
Это должно работать с Sqlalchemy> = 0.6
from sqlalchemy.sql import compiler
from psycopg2.extensions import adapt as sqlescape
# or use the appropiate escape function from your db driver
def compile_query(query):
dialect = query.session.bind.dialect
statement = query.statement
comp = compiler.SQLCompiler(dialect, statement)
comp.compile()
enc = dialect.encoding
params = {}
for k,v in comp.params.iteritems():
if isinstance(v, unicode):
v = v.encode(enc)
params[k] = sqlescape(v)
return (comp.string.encode(enc) % params).decode(enc)
adapt
таким образом. При минимальном вызове подготавливайте () к возвращаемому значению от него каждый раз, предоставляя соединение в качестве аргумента, чтобы он мог выполнять правильное цитирование.
– Alex Gaynor
6 January 2011 в 21:24
prepare
на возвращаемое значение, но кажется, что у него нет этого метода: AttributeError: 'psycopg2._psycopg.AsIs' object has no attribute 'prepare'
. Я использую psycopg2 2.2.1 BTW
– albertov
7 January 2011 в 15:15
Вы можете использовать события из семейства ConnectionEvents : after_cursor_execute
или before_cursor_execute
.
В sqlalchemy UsageRecipes by @zzzeek вы можете найти этот пример:
Profiling
...
@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement,
parameters, context, executemany):
conn.info.setdefault('query_start_time', []).append(time.time())
logger.debug("Start Query: %s" % statement % parameters)
...
Здесь вы можете получить доступ к вашему оператору
Для бэкэнда MySQLdb я изменил внушительный ответ Альбертова (спасибо большое!). Я уверен, что они могут быть объединены, чтобы проверить, было ли comp.positional True, но это немного выходит за рамки этого вопроса.
def compile_query(query):
from sqlalchemy.sql import compiler
from MySQLdb.converters import conversions, escape
dialect = query.session.bind.dialect
statement = query.statement
comp = compiler.SQLCompiler(dialect, statement)
comp.compile()
enc = dialect.encoding
params = []
for k in comp.positiontup:
v = comp.params[k]
if isinstance(v, unicode):
v = v.encode(enc)
params.append( escape(v, conversions) )
return (comp.string.encode(enc) % tuple(params)).decode(enc)
return tuple(params)
работал как шарм! Вы спасли мне бесчисленные часы, чтобы спуститься с чрезвычайно болезненной дороги.
– horcle_buzz
2 November 2015 в 04:00
В следующем решении используется язык выражения SQLAlchemy и работает с SQLAlchemy 1.1. Это решение не смешивает параметры с запросом (в соответствии с запросом оригинального автора), но предоставляет способ использования моделей SQLAlchemy для генерации строк SQL-запросов и словарей параметров для разных диалектов SQL. Пример основан на учебнике http://docs.sqlalchemy.org/en/rel_1_0/core/tutorial.html
. Учитывая класс,
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class foo(Base):
__tablename__ = 'foo'
id = Column(Integer(), primary_key=True)
name = Column(String(80), unique=True)
value = Column(Integer())
мы можем создать оператор запроса с помощью функции select.
from sqlalchemy.sql import select
statement = select([foo.name, foo.value]).where(foo.value > 0)
Далее мы можем скомпилировать оператор в объект запроса.
query = statement.compile()
По умолчанию, оператор компилируется с использованием базовой реализации «named», которая совместима с базами данных SQL, такими как SQLite и Oracle. Если вам нужно указать диалект, такой как PostgreSQL, вы можете сделать
from sqlalchemy.dialects import postgresql
query = statement.compile(dialect=postgresql.dialect())
. Если вы хотите явно указать диалект как SQLite, вы можете изменить paramstyle с «qmark» на «named».
from sqlalchemy.dialects import sqlite
query = statement.compile(dialect=sqlite.dialect(paramstyle="named"))
Из объекта запроса мы можем извлечь строку запроса и параметры запроса
query_str = str(query)
query_params = query.params
и, наконец, выполнить запрос.
conn.execute( query_str, query_params )
Сначала позвольте мне предисловие, сказав, что я предполагаю, что вы делаете это в основном для целей отладки. Я бы не рекомендовал пытаться изменить инструкцию вне свободно распространяемого API SQLAlchemy.
К сожалению, Кажется, это простой способ показать скомпилированный оператор с включенными параметрами запроса. SQLAlchemy фактически не вводит параметры в оператор - они переданы в механизм базы данных в качестве словаря . Это позволяет обработать библиотеку конкретных баз данных с помощью специальных символов и т. П., Чтобы избежать инъекции SQL.
Но вы можете сделать это в двухэтапном процессе достаточно легко. Чтобы получить оператор, вы можете сделать, как вы уже показали, и просто распечатать запрос:
>>> print(query)
SELECT field_1, field_2 FROM table WHERE id=%s;
. Вы можете приблизиться к запросу query.statement, чтобы увидеть имена параметров. (Примечание :id_1
ниже vs %s
выше - на самом деле не проблема в этом очень простом примере, но может быть ключом к более сложному утверждению.)
>>> print(query.statement)
>>> print(query.statement.compile()) # reasonably equivalent, you can also
# pass in a dialect if you want
SELECT field_1, field_2 FROM table WHERE id=:id_1;
Затем вы можете получить фактические значения параметров, получив свойство params
скомпилированного оператора:
>>> print(query.statement.compile().params)
{u'id_1': 1}
Это работало, по крайней мере, на сервере MySQL; Я бы предположил, что это также достаточно общее для PostgreSQL без использования psycopg2
.
SELECT id WHERE date_added <= %s AND date_added >= %s ORDER BY count DESC
IS окончательный запрос. Те %s
отправляются в базу данных по sqlalchemy - sqlalchemy NEVER помещает фактические данные вместо% s
– nosklo
6 January 2011 в 20:51
sqlalchemy.dialects.mysql.mysqldb
, do_executemany()
передает утверждение & amp; параметры отдельно к курсору MySQLdb. yay косвенность!
– cce
6 January 2011 в 23:44
Документация использует literal_binds
для печати запроса q
, включая параметры:
print(q.statement.compile(compile_kwargs={"literal_binds": True}))
, упомянутый выше подход имеет оговорки, что он поддерживается только для базовых типов, таких как ints и strings, и, кроме того, если bindparam () без предварительно заданного значения используется напрямую, он также не сможет выполнить это.
Для бэкэнд postgresql, использующего psycopg2, вы можете прослушивать событие do_execute
, а затем использовать параметры курсора, оператора и типа с принудительным параметром вместе с Cursor.mogrify()
для вставки параметров. Вы можете вернуть True, чтобы предотвратить фактическое выполнение запроса.
import sqlalchemy
class QueryDebugger(object):
def __init__(self, engine, query):
with engine.connect() as connection:
try:
sqlalchemy.event.listen(engine, "do_execute", self.receive_do_execute)
connection.execute(query)
finally:
sqlalchemy.event.remove(engine, "do_execute", self.receive_do_execute)
def receive_do_execute(self, cursor, statement, parameters, context):
self.statement = statement
self.parameters = parameters
self.query = cursor.mogrify(statement, parameters)
# Don't actually execute
return True
Пример использования:
>>> engine = sqlalchemy.create_engine("postgresql://postgres@localhost/test")
>>> metadata = sqlalchemy.MetaData()
>>> users = sqlalchemy.Table('users', metadata, sqlalchemy.Column("_id", sqlalchemy.String, primary_key=True), sqlalchemy.Column("document", sqlalchemy.dialects.postgresql.JSONB))
>>> s = sqlalchemy.select([users.c.document.label("foobar")]).where(users.c.document.contains({"profile": {"iid": "something"}}))
>>> q = QueryDebugger(engine, s)
>>> q.query
'SELECT users.document AS foobar \nFROM users \nWHERE users.document @> \'{"profile": {"iid": "something"}}\''
>>> q.statement
'SELECT users.document AS foobar \nFROM users \nWHERE users.document @> %(document_1)s'
>>> q.parameters
{'document_1': '{"profile": {"iid": "something"}}'}
Я думаю, что .statement, возможно, выполнит трюк: http://docs.sqlalchemy.org/en/latest/orm/query.html?highlight=query
>>> local_session.query(sqlalchemy_declarative.SomeTable.text).statement
<sqlalchemy.sql.annotation.AnnotatedSelect at 0x6c75a20; AnnotatedSelectobject>
>>> x=local_session.query(sqlalchemy_declarative.SomeTable.text).statement
>>> print(x)
SELECT sometable.text
FROM sometable
c = q.statement.compile(...)
, вы можете просто получитьc.params
– Hannele 26 July 2016 в 20:08as_scalar()
-методаQuery
. – Patrick B. 11 July 2018 в 09:10