Skip to content

Instantly share code, notes, and snippets.

@everilae
Last active April 4, 2020 02:24
Show Gist options
  • Save everilae/6d407a94acbc39b30b5e76584d32bcf7 to your computer and use it in GitHub Desktop.
Save everilae/6d407a94acbc39b30b5e76584d32bcf7 to your computer and use it in GitHub Desktop.
Hacky JSON support for SQLAlchemy SQLite dialect, if https://www.sqlite.org/json1.html extension is available.
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.dialects import plugins
from sqlalchemy.dialects.mysql.json import JSONIndexType, JSONPathType
from sqlalchemy.engine import CreateEnginePlugin
from sqlalchemy.sql.elements import BinaryExpression
from sqlalchemy.sql import sqltypes
from sqlalchemy.sql.operators import \
json_getitem_op, \
json_path_getitem_op
__all__ = [
"JSON",
"JsonPlugin"
]
_json_serializer = None
_json_deserializer = None
class JSON(sqltypes.JSON):
class Comparator(sqltypes.JSON.Comparator):
def _setup_getitem(self, index):
operator, index, _ = super()._setup_getitem(index)
# https://www.sqlite.org/json1.html#jex
# "the SQL datatype of the result is NULL for a JSON null, INTEGER
# or REAL for a JSON numeric value, an INTEGER zero for a JSON false
# value, an INTEGER one for a JSON true value, the dequoted text for
# a JSON string value, and a text representation for JSON object and
# array values. If there are multiple path arguments (P1, P2, and so
# forth) then this routine returns SQLite text which is a
# well-formed JSON array holding the various values."
return operator, index, sqltypes.NullType()
comparator_factory = Comparator
@compiles(JSON, "sqlite")
@compiles(sqltypes.JSON, "sqlite")
def compile_json_type(element, compiler, **kw):
return "JSON"
@compiles(BinaryExpression, "sqlite")
def compile_binary(binary, compiler, override_operator=None, **kw):
operator = override_operator or binary.operator
if operator is json_getitem_op:
return visit_json_getitem_op_binary(
compiler, binary, operator, override_operator=override_operator,
**kw)
if operator is json_path_getitem_op:
return visit_json_path_getitem_op_binary(
compiler, binary, operator, override_operator=override_operator,
**kw)
return compiler.process(binary, override_operator=override_operator, **kw)
def visit_json_getitem_op_binary(compiler, binary, operator, **kw):
return "JSON_EXTRACT(%s, %s)" % (
compiler.process(binary.left),
compiler.process(binary.right))
def visit_json_path_getitem_op_binary(compiler, binary, operator, **kw):
return "JSON_EXTRACT(%s, %s)" % (
compiler.process(binary.left),
compiler.process(binary.right))
def monkeypatch_dialect(dialect):
if not hasattr(dialect, "_json_serializer"):
dialect._json_serializer = _json_serializer
if not hasattr(dialect, "_json_deserializer"):
dialect._json_deserializer = _json_deserializer
if sqltypes.JSON not in dialect.colspecs:
dialect.colspecs = dialect.colspecs.copy()
dialect.colspecs[sqltypes.JSON] = JSON
dialect.colspecs[sqltypes.JSON.JSONIndexType] = JSONIndexType
dialect.colspecs[sqltypes.JSON.JSONPathType] = JSONPathType
if "json" not in dialect.ischema_names:
dialect.ischema_names = dialect.ischema_names.copy()
dialect.ischema_names["json"] = JSON
class JsonPlugin(CreateEnginePlugin):
def engine_created(self, engine):
monkeypatch_dialect(engine.dialect)
plugins.register("jsonplugin", "sqlite_json", "JsonPlugin")
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, create_engine, func
from sqlite_json import JSON
engine = create_engine("sqlite:///?plugin=jsonplugin", echo=True)
Base = declarative_base()
Base.metadata.bind = engine
Session = sessionmaker()
class Foo(Base):
__tablename__ = "foo"
id = Column(Integer, primary_key=True)
data = Column(JSON)
Base.metadata.create_all()
session = Session()
session.add(Foo(data={"a": 1, "b": {"c": 2}}))
session.commit()
print(session.query(Foo).first().data)
print(session.query(func.json_extract(Foo.data, "$.a")).scalar())
print(session.query(Foo.data["a"]).scalar())
print(session.query(Foo.data["b", "c"]).scalar())
@everilae
Copy link
Author

everilae commented Mar 9, 2018

Store as sqlite_json module in your Python PATH. The plugin registry has to find it that way.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment