Merge pull request #2974 from bassettb/bassettb/unquote-user

add unquote_user option

playhouse/db_url.py

@@ -39,7 +39,7 @@ def register_database(db_class, *names):
     for name in names:
         schemes[name] = db_class
 
-def parseresult_to_dict(parsed, unquote_password=False):
+def parseresult_to_dict(parsed, unquote_password=False, unquote_user=False):
 
     # urlparse in python 2.6 is broken so query will be empty and instead
     # appended to path complete with '?'
@@ -49,6 +49,8 @@ def parseresult_to_dict(parsed, unquote_password=False):
     connect_kwargs = {'database': path}
     if parsed.username:
         connect_kwargs['user'] = parsed.username
+        if unquote_user:
+            connect_kwargs['user'] = unquote(connect_kwargs['user'])
     if parsed.password:
         connect_kwargs['password'] = parsed.password
         if unquote_password:
@@ -85,13 +87,13 @@ def parseresult_to_dict(parsed, unquote_password=False):
 
     return connect_kwargs
 
-def parse(url, unquote_password=False):
+def parse(url, unquote_password=False, unquote_user=False):
     parsed = urlparse(url)
-    return parseresult_to_dict(parsed, unquote_password)
+    return parseresult_to_dict(parsed, unquote_password, unquote_user)
 
-def connect(url, unquote_password=False, **connect_params):
+def connect(url, unquote_password=False, unquote_user=False, **connect_params):
     parsed = urlparse(url)
-    connect_kwargs = parseresult_to_dict(parsed, unquote_password)
+    connect_kwargs = parseresult_to_dict(parsed, unquote_password, unquote_user)
     connect_kwargs.update(connect_params)
     database_class = schemes.get(parsed.scheme)
 

tests/db_url.py

@@ -29,22 +29,31 @@ def test_db_url_parse(self):
         self.assertEqual(cfg['baz'], '3.4.5')
         self.assertEqual(cfg['boolz'], False)
 
-    def test_db_url_quoted_password(self):
-        # By default, the password is not unescaped.
-        cfg = parse('mysql://usr:pwd%23%20@hst:123/db')
-        self.assertEqual(cfg['user'], 'usr')
-        self.assertEqual(cfg['passwd'], 'pwd%23%20')
+    def test_db_url_no_unquoting(self):
+        # By default, neither user nor password is not unescaped.
+        cfg = parse('mysql://usr%40example.com:pwd%23@hst:123/db')
+        self.assertEqual(cfg['user'], 'usr%40example.com')
+        self.assertEqual(cfg['passwd'], 'pwd%23')
         self.assertEqual(cfg['host'], 'hst')
         self.assertEqual(cfg['database'], 'db')
         self.assertEqual(cfg['port'], 123)
 
+    def test_db_url_quoted_password(self):
         cfg = parse('mysql://usr:pwd%23%20@hst:123/db', unquote_password=True)
         self.assertEqual(cfg['user'], 'usr')
         self.assertEqual(cfg['passwd'], 'pwd# ')
         self.assertEqual(cfg['host'], 'hst')
         self.assertEqual(cfg['database'], 'db')
         self.assertEqual(cfg['port'], 123)
 
+    def test_db_url_quoted_user(self):
+        cfg = parse('mysql://usr%40example.com:p%40sswd@hst:123/db', unquote_user=True)
+        self.assertEqual(cfg['user'], '[email protected]')
+        self.assertEqual(cfg['passwd'], 'p%40sswd')
+        self.assertEqual(cfg['host'], 'hst')
+        self.assertEqual(cfg['database'], 'db')
+        self.assertEqual(cfg['port'], 123)
+
     def test_db_url(self):
         db = connect('sqlite:///:memory:')
         self.assertTrue(isinstance(db, SqliteDatabase))