Add support for LONGTEXT, MEDIUMTEXT to JsonEncodedType

These types has the capacity of 2^32 and 2^24 bytes respectively
in mysql and are needed in various use-cases.

Change-Id: I498803eb0f294c0402666ee3911374a5d0399b82
Closes-Bug: #1599800
Related-Bug: https://bugs.launchpad.net/mistral/+bug/1438101
Related-Bug: https://bugs.launchpad.net/ironic/+bug/1596421

2 files changed, 38 insertions, 0 deletions

diff --git a/oslo_db/sqlalchemy/types.py b/oslo_db/sqlalchemy/types.py
index 477ed3f..a6f8acb 100644
--- a/oslo_db/sqlalchemy/types.py
+++ b/oslo_db/sqlalchemy/types.py
@@ -13,6 +13,7 @@
 import json
 
 from sqlalchemy.types import TypeDecorator, Text
+from sqlalchemy.dialects import mysql
 
 
 class JsonEncodedType(TypeDecorator):
@@ -20,6 +21,18 @@ class JsonEncodedType(TypeDecorator):
     type = None
     impl = Text
 
+    def __init__(self, mysql_as_long=False, mysql_as_medium=False):
+        super(JsonEncodedType, self).__init__()
+
+        if mysql_as_long and mysql_as_medium:
+            raise TypeError("mysql_as_long and mysql_as_medium are mutually "
+                            "exclusive")
+
+        if mysql_as_long:
+            self.impl = Text().with_variant(mysql.LONGTEXT(), 'mysql')
+        elif mysql_as_medium:
+            self.impl = Text().with_variant(mysql.MEDIUMTEXT(), 'mysql')
+
     def process_bind_param(self, value, dialect):
         if value is None:
             if self.type is not None:
diff --git a/oslo_db/tests/sqlalchemy/test_types.py b/oslo_db/tests/sqlalchemy/test_types.py
index c8abbe1..b3787a2 100644
--- a/oslo_db/tests/sqlalchemy/test_types.py
+++ b/oslo_db/tests/sqlalchemy/test_types.py
@@ -13,6 +13,7 @@
 """Tests for JSON SQLAlchemy types."""
 
 from sqlalchemy import Column, Integer
+from sqlalchemy.dialects import mysql
 from sqlalchemy.ext.declarative import declarative_base
 
 from oslo_db import exception as db_exc
@@ -84,3 +85,27 @@ class JsonTypesTestCase(test_base.DbTestCase):
                 JsonTable(id=i, json=test).save(self.session)
             obj = self.session.query(JsonTable).filter_by(id=i).one()
             self.assertEqual(test, obj.json)
+
+    def test_mysql_variants(self):
+        self.assertEqual(
+            str(
+                types.JsonEncodedDict(mysql_as_long=True).compile(
+                    dialect=mysql.dialect())
+            ),
+            "LONGTEXT"
+        )
+
+        self.assertEqual(
+            str(
+                types.JsonEncodedDict(mysql_as_medium=True).compile(
+                    dialect=mysql.dialect())
+            ),
+            "MEDIUMTEXT"
+        )
+
+        self.assertRaises(
+            TypeError,
+            lambda: types.JsonEncodedDict(
+                mysql_as_long=True,
+                mysql_as_medium=True)
+        )