engine = create_engine(engineUrl, convert_unicode=True, poolclass=NullPool)
def jsonExtract(item, key):
j = json.loads(item)
value = j.get(key)
return value
@sqlalchemy.event.listens_for(engine, "begin")
def do_begin(conn):
if conn.dialect.name == 'sqlite':
conn.connection.create_function('jsonExtract', 2, jsonExtract)
class SqliteJsonElement(FunctionElement):
type = types.Text()
name = "JsonElement"
@property
def astext(self):
return self
@compiles(SqliteJsonElement)
def compileSqliteJsonElement(element, compiler, **kw):
return "jsonExtract(%s)" % compiler.process(element.clauses)
class SqliteJson(TypeDecorator):
impl = types.Text
def process_bind_param(self, value, dialect):
return json.dumps(value)
def process_result_value(self, value, dialect):
return json.loads(value)
class comparator_factory(sqltypes.Concatenable.Comparator):
def __getitem__(self, other):
"""Get the value at a given key."""
return SqliteJsonElement(self.expr, other)
class SqliteCompatiblePostgresOptimizedJson(TypeDecorator):
# See: http://docs.sqlalchemy.org/en/latest/core/custom_types.html
impl = None
def load_dialect_impl(self, dialect):
dialectTypeMap = {
'sqlite': SqliteJson,
'postgresql': JSONB
}
self.impl = dialect.type_descriptor(dialectTypeMap[dialect.name])
return self.impl
def process_bind_param(self, value, dialect):
return value
def process_result_value(self, value, dialect):
return value
@property
def comparator_factory(self):
"""express comparison behavior in terms of the base type"""
return self.impl.comparator_factory
Last active
August 11, 2017 10:37
Sqlalchemy Sqlite JSON Type Supporting Index Queries
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment