Sqlite json

Added support for SQLite's json functionality via the new
SQLite implementation for :class:`.sqltypes.JSON`, :class:`.sqlite.JSON`.
The name used for the type is ``JSON``, following an example found at
SQLite's own documentation. Pull request courtesy Ilja Everilä.

Fixes: #3850
Change-Id: I3d2714fb8655343a99d13dc751b16b93d05d7dda
Pull-request: https://github.com/zzzeek/sqlalchemy/pull/434

1 files changed, 62 insertions, 0 deletions

diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index d2d563208..5e2535b30 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -219,6 +219,68 @@ class TestTypes(fixtures.TestBase, AssertsExecutionResults):
                 isinstance(bindproc(util.u('some string')), util.text_type)
 
 
+class JSONTest(fixtures.TestBase):
+
+    __requires__ = ('json_type', )
+    __only_on__ = 'sqlite'
+
+    @testing.provide_metadata
+    @testing.requires.reflects_json_type
+    def test_reflection(self):
+        Table(
+            'json_test', self.metadata,
+            Column('foo', sqlite.JSON)
+        )
+        self.metadata.create_all()
+
+        reflected = Table('json_test', MetaData(), autoload_with=testing.db)
+        is_(reflected.c.foo.type._type_affinity, sqltypes.JSON)
+        assert isinstance(reflected.c.foo.type, sqlite.JSON)
+
+    @testing.provide_metadata
+    def test_rudimentary_roundtrip(self):
+        sqlite_json = Table(
+            'json_test', self.metadata,
+            Column('foo', sqlite.JSON)
+        )
+
+        self.metadata.create_all()
+
+        value = {
+            'json': {'foo': 'bar'},
+            'recs': ['one', 'two']
+        }
+
+        with testing.db.connect() as conn:
+            conn.execute(sqlite_json.insert(), foo=value)
+
+            eq_(
+                conn.scalar(select([sqlite_json.c.foo])),
+                value
+            )
+
+    @testing.provide_metadata
+    def test_extract_subobject(self):
+        sqlite_json = Table(
+            'json_test', self.metadata,
+            Column('foo', sqlite.JSON)
+        )
+
+        self.metadata.create_all()
+
+        value = {
+            'json': {'foo': 'bar'},
+        }
+
+        with testing.db.connect() as conn:
+            conn.execute(sqlite_json.insert(), foo=value)
+
+            eq_(
+                conn.scalar(select([sqlite_json.c.foo['json']])),
+                value['json']
+            )
+
+
 class DateTimeTest(fixtures.TestBase, AssertsCompiledSQL):
 
     def test_time_microseconds(self):