Skip to content

Instantly share code, notes, and snippets.

@GermanFilipp
Last active June 12, 2020 11:30
Show Gist options
  • Save GermanFilipp/7d7eb0a8f1fced9291da370e908157b5 to your computer and use it in GitHub Desktop.
Save GermanFilipp/7d7eb0a8f1fced9291da370e908157b5 to your computer and use it in GitHub Desktop.
from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.sql.sqltypes import String, DateTime, NullType
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):
return str(value)
if not isinstance(value, str):
value = str(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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment