Merge ddef74f into 2f1f679

commcare_export/cli.py

@@ -209,6 +209,12 @@ def _get_writer(output_format, output, strict_types):
     elif output_format == 'sql':
         # Output should be a connection URL
         # Writer had bizarre issues so we use a full connection instead of passing in a URL or engine
+        if output.startswith('mysql'):
+            charset_split = output.split('charset=')
+            if len(charset_split) > 1 and charset_split[1] != 'utf8mb4':
+                raise Exception(f"The charset '{charset_split[1]}' might cause problems with the export. "
+                                f"It is recommended that you use 'utf8mb4' instead.")
+
         return writers.SqlTableWriter(output, strict_types)
     else:
         raise Exception("Unknown output format: {}".format(output_format))

commcare_export/writers.py

@@ -262,7 +262,7 @@ class SqlMixin(object):
 
     def __init__(self, db_url, poolclass=None, engine=None):
         self.db_url = db_url
-        self.collation = 'utf8_bin' if 'mysql' in db_url else None
+        self.collation = 'utf8mb4_unicode_ci' if 'mysql' in db_url else None
         self.engine = engine or sqlalchemy.create_engine(db_url, poolclass=poolclass)
 
     def __enter__(self):

tests/conftest.py

@@ -66,7 +66,7 @@ def tear_down():
         'admin_db': 'postgres'
     }, marks=pytest.mark.postgres),
     pytest.param({
-        'url': '{}%s?charset=utf8'.format(mysql_base),
+        'url': '{}%s?charset=utf8mb4'.format(mysql_base),
     }, marks=pytest.mark.mysql),
     pytest.param({
         'url': '{}%s?driver=ODBC+Driver+17+for+SQL+Server'.format(mssql_base),

tests/test_writers.py

@@ -439,6 +439,16 @@ def test_mssql_nvarchar_length_downsize(self, writer):
             result = self._get_column_lengths(writer.connection, 'mssql_nvarchar_length_downsize')
             assert result['some_data'] == ('some_data', 'nvarchar', -1)
 
+    def test_big_lump_of_poo(self, writer):
+        with writer:
+            writer.write_table(TableSpec(**{
+                'name': 'foo_with_emoji',
+                'headings': ['id', 'fun_to_be_had'],
+                'rows': [
+                    ['A steaming poo', '💩'],
+                    ['2020', '😷'],
+                ],
+            }))
 
     def _get_column_lengths(self, connection, table_name):
         return {