mirror of
https://github.com/psycopg/psycopg2.git
synced 2025-03-03 07:45:45 +03:00
Merge branch 'parse-dsn'
This commit is contained in:
commit
14246a5689
1
NEWS
1
NEWS
|
@ -6,6 +6,7 @@ What's new in psycopg 2.7
|
|||
|
||||
New features:
|
||||
|
||||
- Added `~psycopg2.extensions.parse_dsn()` function (:ticket:`#321`).
|
||||
- Added `~psycopg2.__libpq_version__` and
|
||||
`~psycopg2.extensions.libpq_version()` to inspect the version of the
|
||||
``libpq`` library the module was compiled/loaded with
|
||||
|
|
|
@ -12,6 +12,17 @@
|
|||
The module contains a few objects and function extending the minimum set of
|
||||
functionalities defined by the |DBAPI|_.
|
||||
|
||||
.. function:: parse_dsn(dsn)
|
||||
|
||||
Parse connection string into a dictionary of keywords and values.
|
||||
|
||||
Uses libpq's ``PQconninfoParse`` to parse the string according to
|
||||
accepted format(s) and check for supported keywords.
|
||||
|
||||
Example::
|
||||
|
||||
>>> psycopg2.extensions.parse_dsn('dbname=test user=postgres password=secret')
|
||||
{'password': 'secret', 'user': 'postgres', 'dbname': 'test'}
|
||||
|
||||
.. class:: connection(dsn, async=False)
|
||||
|
||||
|
|
|
@ -78,6 +78,7 @@ The module interface respects the standard defined in the |DBAPI|_.
|
|||
|
||||
.. seealso::
|
||||
|
||||
- `~psycopg2.extensions.parse_dsn`
|
||||
- libpq `connection string syntax`__
|
||||
- libpq supported `connection parameters`__
|
||||
- libpq supported `environment variables`__
|
||||
|
@ -91,7 +92,6 @@ The module interface respects the standard defined in the |DBAPI|_.
|
|||
The parameters *connection_factory* and *async* are Psycopg extensions
|
||||
to the |DBAPI|.
|
||||
|
||||
|
||||
.. data:: apilevel
|
||||
|
||||
String constant stating the supported DB API level. For `psycopg2` is
|
||||
|
|
|
@ -56,7 +56,7 @@ try:
|
|||
except ImportError:
|
||||
pass
|
||||
|
||||
from psycopg2._psycopg import adapt, adapters, encodings, connection, cursor, lobject, Xid, libpq_version
|
||||
from psycopg2._psycopg import adapt, adapters, encodings, connection, cursor, lobject, Xid, libpq_version, parse_dsn
|
||||
from psycopg2._psycopg import string_types, binary_types, new_type, new_array_type, register_type
|
||||
from psycopg2._psycopg import ISQLQuote, Notify, Diagnostics, Column
|
||||
|
||||
|
|
|
@ -112,6 +112,59 @@ psyco_connect(PyObject *self, PyObject *args, PyObject *keywds)
|
|||
return conn;
|
||||
}
|
||||
|
||||
#define psyco_parse_dsn_doc "parse_dsn(dsn) -> dict"
|
||||
|
||||
static PyObject *
|
||||
psyco_parse_dsn(PyObject *self, PyObject *args, PyObject *kwargs)
|
||||
{
|
||||
char *err = NULL;
|
||||
PQconninfoOption *options = NULL, *o;
|
||||
PyObject *dict = NULL, *res = NULL, *dsn;
|
||||
|
||||
static char *kwlist[] = {"dsn", NULL};
|
||||
if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O", kwlist, &dsn)) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
Py_INCREF(dsn); /* for ensure_bytes */
|
||||
if (!(dsn = psycopg_ensure_bytes(dsn))) { goto exit; }
|
||||
|
||||
options = PQconninfoParse(Bytes_AS_STRING(dsn), &err);
|
||||
if (options == NULL) {
|
||||
if (err != NULL) {
|
||||
PyErr_Format(ProgrammingError, "error parsing the dsn: %s", err);
|
||||
PQfreemem(err);
|
||||
} else {
|
||||
PyErr_SetString(OperationalError, "PQconninfoParse() failed");
|
||||
}
|
||||
goto exit;
|
||||
}
|
||||
|
||||
if (!(dict = PyDict_New())) { goto exit; }
|
||||
for (o = options; o->keyword != NULL; o++) {
|
||||
if (o->val != NULL) {
|
||||
PyObject *value;
|
||||
if (!(value = Text_FromUTF8(o->val))) { goto exit; }
|
||||
if (PyDict_SetItemString(dict, o->keyword, value) != 0) {
|
||||
Py_DECREF(value);
|
||||
goto exit;
|
||||
}
|
||||
Py_DECREF(value);
|
||||
}
|
||||
}
|
||||
|
||||
/* success */
|
||||
res = dict;
|
||||
dict = NULL;
|
||||
|
||||
exit:
|
||||
PQconninfoFree(options); /* safe on null */
|
||||
Py_XDECREF(dict);
|
||||
Py_XDECREF(dsn);
|
||||
|
||||
return res;
|
||||
}
|
||||
|
||||
/** type registration **/
|
||||
#define psyco_register_type_doc \
|
||||
"register_type(obj, conn_or_curs) -> None -- register obj with psycopg type system\n\n" \
|
||||
|
@ -708,6 +761,8 @@ error:
|
|||
static PyMethodDef psycopgMethods[] = {
|
||||
{"_connect", (PyCFunction)psyco_connect,
|
||||
METH_VARARGS|METH_KEYWORDS, psyco_connect_doc},
|
||||
{"parse_dsn", (PyCFunction)psyco_parse_dsn,
|
||||
METH_VARARGS|METH_KEYWORDS, psyco_parse_dsn_doc},
|
||||
{"adapt", (PyCFunction)psyco_microprotocols_adapt,
|
||||
METH_VARARGS, psyco_microprotocols_adapt_doc},
|
||||
|
||||
|
|
|
@ -23,6 +23,7 @@
|
|||
# License for more details.
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import threading
|
||||
from operator import attrgetter
|
||||
|
@ -33,7 +34,7 @@ import psycopg2.errorcodes
|
|||
import psycopg2.extensions
|
||||
|
||||
from testutils import unittest, decorate_all_tests, skip_if_no_superuser
|
||||
from testutils import skip_before_postgres, skip_after_postgres
|
||||
from testutils import skip_before_postgres, skip_after_postgres, skip_before_libpq
|
||||
from testutils import ConnectingTestCase, skip_if_tpc_disabled
|
||||
from testutils import skip_if_windows
|
||||
from testconfig import dsn, dbname
|
||||
|
@ -308,6 +309,78 @@ class ConnectionTests(ConnectingTestCase):
|
|||
self.assert_('foobar' not in c.dsn, "password was not obscured")
|
||||
|
||||
|
||||
class ParseDsnTestCase(ConnectingTestCase):
|
||||
def test_parse_dsn(self):
|
||||
from psycopg2 import ProgrammingError
|
||||
from psycopg2.extensions import parse_dsn
|
||||
|
||||
self.assertEqual(parse_dsn('dbname=test user=tester password=secret'),
|
||||
dict(user='tester', password='secret', dbname='test'),
|
||||
"simple DSN parsed")
|
||||
|
||||
self.assertRaises(ProgrammingError, parse_dsn,
|
||||
"dbname=test 2 user=tester password=secret")
|
||||
|
||||
self.assertEqual(parse_dsn("dbname='test 2' user=tester password=secret"),
|
||||
dict(user='tester', password='secret', dbname='test 2'),
|
||||
"DSN with quoting parsed")
|
||||
|
||||
# Can't really use assertRaisesRegexp() here since we need to
|
||||
# make sure that secret is *not* exposed in the error messgage
|
||||
# (and it also requires python >= 2.7).
|
||||
raised = False
|
||||
try:
|
||||
# unterminated quote after dbname:
|
||||
parse_dsn("dbname='test 2 user=tester password=secret")
|
||||
except ProgrammingError, e:
|
||||
raised = True
|
||||
self.assertTrue(str(e).find('secret') < 0,
|
||||
"DSN was not exposed in error message")
|
||||
except e:
|
||||
self.fail("unexpected error condition: " + repr(e))
|
||||
self.assertTrue(raised, "ProgrammingError raised due to invalid DSN")
|
||||
|
||||
@skip_before_libpq(9, 2)
|
||||
def test_parse_dsn_uri(self):
|
||||
from psycopg2.extensions import parse_dsn
|
||||
|
||||
self.assertEqual(parse_dsn('postgresql://tester:secret@/test'),
|
||||
dict(user='tester', password='secret', dbname='test'),
|
||||
"valid URI dsn parsed")
|
||||
|
||||
raised = False
|
||||
try:
|
||||
# extra '=' after port value
|
||||
parse_dsn(dsn='postgresql://tester:secret@/test?port=1111=x')
|
||||
except psycopg2.ProgrammingError, e:
|
||||
raised = True
|
||||
self.assertTrue(str(e).find('secret') < 0,
|
||||
"URI was not exposed in error message")
|
||||
except e:
|
||||
self.fail("unexpected error condition: " + repr(e))
|
||||
self.assertTrue(raised, "ProgrammingError raised due to invalid URI")
|
||||
|
||||
def test_unicode_value(self):
|
||||
from psycopg2.extensions import parse_dsn
|
||||
snowman = u"\u2603"
|
||||
d = parse_dsn('dbname=' + snowman)
|
||||
if sys.version_info[0] < 3:
|
||||
self.assertEqual(d['dbname'], snowman.encode('utf8'))
|
||||
else:
|
||||
self.assertEqual(d['dbname'], snowman)
|
||||
|
||||
def test_unicode_key(self):
|
||||
from psycopg2.extensions import parse_dsn
|
||||
snowman = u"\u2603"
|
||||
self.assertRaises(psycopg2.ProgrammingError, parse_dsn,
|
||||
snowman + '=' + snowman)
|
||||
|
||||
def test_bad_param(self):
|
||||
from psycopg2.extensions import parse_dsn
|
||||
self.assertRaises(TypeError, parse_dsn, None)
|
||||
self.assertRaises(TypeError, parse_dsn, 42)
|
||||
|
||||
|
||||
class IsolationLevelsTestCase(ConnectingTestCase):
|
||||
|
||||
def setUp(self):
|
||||
|
|
Loading…
Reference in New Issue
Block a user