Request: SQLite json1 ext support

[Shengjing Zhu]

sqlalchemy.types.JSON only supports PostgreSQL and MySQL. But since SQLite 3.9, it has json support, https://sqlite.org/json1.html

I hope sqlalchemy can support json1 too.

[Mike Bayer]

im not sure I have the resources to support a SQLIte JSON type directly right now as I generally do not support SQLite extensions. However, SQLAlchemy's type system is fully extensible so I will put up a recipe soon.

[Mike Bayer]

hm, it seems those extensions come pre-installed, OK. it's just SQL functions so there's not much here.

[Mike Bayer]

Here's a recipe. A real version of this feature would be sent as a PR that looks a lot like the MySQL JSON feature, as the syntax here seems to work mostly the same way, except I'm not sure MySQL's json_extract() reutrns non-JSON values the way this one does (maybe it does).

from sqlalchemy import TypeDecorator, String, func
from sqlalchemy.types import NullType
import json


class SQLiteJson(TypeDecorator):
    impl = String

    class Comparator(String.Comparator):
        def __getitem__(self, index):
            if isinstance(index, tuple):
                index = "$%s" % (
                    "".join([
                        "[%s]" % elem if isinstance(elem, int)
                        else '."%s"' % elem for elem in index
                    ])
                )
            elif isinstance(index, int):
                index = "$[%s]" % index
            else:
                index = '$."%s"' % index

            # json_extract does not appear to return JSON sub-elements
            # which is weird.
            return func.json_extract(self.expr, index, type_=NullType)

    comparator_factory = Comparator

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = json.dumps(value)
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = json.loads(value)
        return value

from sqlalchemy import create_engine, Table, MetaData, Column, Integer, select

e = create_engine("sqlite://", echo='debug')

m = MetaData()
t = Table(
    't', m, Column('id', Integer, primary_key=True),
    Column('data', SQLiteJson)

)
m.create_all(e)

with e.connect() as conn:
    conn.execute(
        t.insert(), data={"x": {"y": "two", "z": [1, 2, 3]}}
    )

    value = conn.scalar(
        select([t.c.data[('x', 'y')]]).where(t.c.data[('x', 'y')] == 'two')
    )
    print value

[Timothée Jeannin]

Would be very nice to have support for the JSON type in SQLite databases !