```python 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 ```