@classmethod
def connect(
- cls,
- conninfo: Optional[str] = None,
- *,
- autocommit: bool = False,
- **kwargs: Any,
+ cls, conninfo: str = "", *, autocommit: bool = False, **kwargs: Any
) -> "Connection":
- if conninfo is None and not kwargs:
- raise TypeError("missing conninfo and not parameters specified")
- conninfo = make_conninfo(conninfo or "", **kwargs)
+ """
+ Connect to a database server and return a new `Connection` instance.
+
+ TODO: connection_timeout to be implemented.
+ """
+
+ conninfo = make_conninfo(conninfo, **kwargs)
gen = connect(conninfo)
pgconn = cls.wait(gen)
conn = cls(pgconn)
@classmethod
async def connect(
- cls,
- conninfo: Optional[str] = None,
- *,
- autocommit: bool = False,
- **kwargs: Any,
+ cls, conninfo: str = "", *, autocommit: bool = False, **kwargs: Any
) -> "AsyncConnection":
- if conninfo is None and not kwargs:
- raise TypeError("missing conninfo and not parameters specified")
- conninfo = make_conninfo(conninfo or "", **kwargs)
+ """`asyncio` version of `~Connection.connect()`."""
+ conninfo = make_conninfo(conninfo, **kwargs)
gen = connect(conninfo)
pgconn = await cls.wait(gen)
conn = cls(pgconn)
@pytest.mark.parametrize(
- "testdsn, kwargs, want",
+ "args, kwargs, want",
[
- ("", {}, ""),
- ("host=foo user=bar", {}, "host=foo user=bar"),
- ("host=foo", {"user": "baz"}, "host=foo user=baz"),
+ ((), {}, ""),
+ (("",), {}, ""),
+ (("host=foo user=bar",), {}, "host=foo user=bar"),
+ (("host=foo",), {"user": "baz"}, "host=foo user=baz"),
(
- "host=foo port=5432",
+ ("host=foo port=5432",),
{"host": "qux", "user": "joe"},
"host=qux user=joe port=5432",
),
- ("host=foo", {"user": None}, "host=foo"),
+ (("host=foo",), {"user": None}, "host=foo"),
],
)
-def test_connect_args(monkeypatch, pgconn, testdsn, kwargs, want):
+def test_connect_args(monkeypatch, pgconn, args, kwargs, want):
the_conninfo = None
def fake_connect(conninfo):
yield
monkeypatch.setattr(psycopg3.connection, "connect", fake_connect)
- psycopg3.Connection.connect(testdsn, **kwargs)
+ psycopg3.Connection.connect(*args, **kwargs)
assert conninfo_to_dict(the_conninfo) == conninfo_to_dict(want)
@pytest.mark.parametrize(
- "args, kwargs", [((), {}), (("", ""), {}), ((), {"nosuchparam": 42})]
+ "args, kwargs",
+ [
+ (("host=foo", "host=bar"), {}),
+ (("", ""), {}),
+ ((), {"nosuchparam": 42}),
+ ],
)
def test_connect_badargs(monkeypatch, pgconn, args, kwargs):
def fake_connect(conninfo):
@pytest.mark.parametrize(
- "testdsn, kwargs, want",
+ "args, kwargs, want",
[
- ("", {}, ""),
- ("host=foo user=bar", {}, "host=foo user=bar"),
- ("host=foo", {"user": "baz"}, "host=foo user=baz"),
+ ((), {}, ""),
+ (("",), {}, ""),
+ (("host=foo user=bar",), {}, "host=foo user=bar"),
+ (("host=foo",), {"user": "baz"}, "host=foo user=baz"),
(
- "host=foo port=5432",
+ ("host=foo port=5432",),
{"host": "qux", "user": "joe"},
"host=qux user=joe port=5432",
),
- ("host=foo", {"user": None}, "host=foo"),
+ (("host=foo",), {"user": None}, "host=foo"),
],
)
-async def test_connect_args(monkeypatch, pgconn, testdsn, kwargs, want):
+async def test_connect_args(monkeypatch, pgconn, args, kwargs, want):
the_conninfo = None
def fake_connect(conninfo):
yield
monkeypatch.setattr(psycopg3.connection, "connect", fake_connect)
- await psycopg3.AsyncConnection.connect(testdsn, **kwargs)
+ await psycopg3.AsyncConnection.connect(*args, **kwargs)
assert conninfo_to_dict(the_conninfo) == conninfo_to_dict(want)
@pytest.mark.parametrize(
- "args, kwargs", [((), {}), (("", ""), {}), ((), {"nosuchparam": 42})]
+ "args, kwargs",
+ [
+ (("host=foo", "host=bar"), {}),
+ (("", ""), {}),
+ ((), {"nosuchparam": 42}),
+ ],
)
async def test_connect_badargs(monkeypatch, pgconn, args, kwargs):
def fake_connect(conninfo):
@pytest.mark.parametrize(
- "testdsn, kwargs, want",
+ "args, kwargs, want",
[
- ("", {}, ""),
- ("host=foo user=bar", {}, "host=foo user=bar"),
- ("host=foo", {"user": "baz"}, "host=foo user=baz"),
+ ((), {}, ""),
+ (("",), {}, ""),
+ (("host=foo user=bar",), {}, "host=foo user=bar"),
+ (("host=foo",), {"user": "baz"}, "host=foo user=baz"),
(
- "host=foo port=5432",
+ ("host=foo port=5432",),
{"host": "qux", "user": "joe"},
"host=qux user=joe port=5432",
),
- ("host=foo", {"user": None}, "host=foo"),
+ (("host=foo",), {"user": None}, "host=foo"),
],
)
-def test_connect_args(monkeypatch, pgconn, testdsn, kwargs, want):
+def test_connect_args(monkeypatch, pgconn, args, kwargs, want):
the_conninfo = None
def fake_connect(conninfo):
yield
monkeypatch.setattr(psycopg3.connection, "connect", fake_connect)
- psycopg3.connect(testdsn, **kwargs)
+ psycopg3.connect(*args, **kwargs)
assert conninfo_to_dict(the_conninfo) == conninfo_to_dict(want)
@pytest.mark.parametrize(
- "args, kwargs", [((), {}), (("", ""), {}), ((), {"nosuchparam": 42})]
+ "args, kwargs",
+ [
+ (("host=foo", "host=bar"), {}),
+ (("", ""), {}),
+ ((), {"nosuchparam": 42}),
+ ],
)
def test_connect_badargs(monkeypatch, pgconn, args, kwargs):
def fake_connect(conninfo):