PublicArchive/peewee
2
0
Fork 0
mirror of https://github.com/coleifer/peewee.git synced 2026-05-06 07:56:41 -04:00
Code Issues Packages Projects Releases Wiki Activity

unquote_user

Browse Source
This commit is contained in:
Bryan Bassett
2025-02-20 13:00:23 -05:00
parent 2fd641106a
commit 7f45e4eefb
2 changed files with 21 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
playhouse/db_url.py
+7 -5
View File
@@ -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
+14 -5
View File
@@ -29,15 +29,16 @@ class TestDBUrl(BaseTestCase):
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# ')
@@ -45,6 +46,14 @@ class TestDBUrl(BaseTestCase):
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'], 'usr@example.com')
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))