from functools import lru_cache
import re
from typing import Any
+from typing import cast
from typing import List
from typing import Optional
from typing import Tuple
+from typing import TYPE_CHECKING
from typing import Union
from . import array as _array
Tuple[None, None],
Tuple[Tuple[Optional[str], ...], Tuple[Optional[int], ...]],
]:
- hosts = ports = None
+ hosts: Optional[Tuple[Optional[str], ...]] = None
+ ports_str: Union[str, Tuple[Optional[str], ...], None] = None
integrated_multihost = False
if "host" in url.query:
if isinstance(url.query["host"], (list, tuple)):
integrated_multihost = True
- hosts, ports = zip(
+ hosts, ports_str = zip(
*[
token.split(":") if ":" in token else (token, None)
for token in url.query["host"]
if host_port_match:
integrated_multihost = True
h, p = host_port_match.group(1, 2)
+ if TYPE_CHECKING:
+ assert isinstance(h, str)
+ assert isinstance(p, str)
hosts = (h,)
- ports = (p,) if p else (None,)
+ ports_str = cast(
+ "Tuple[Optional[str], ...]", (p,) if p else (None,)
+ )
if "port" in url.query:
if integrated_multihost:
'"host=h1:p1&host=h2:p2&host=h3:p3" separately'
)
if isinstance(url.query["port"], (list, tuple)):
- ports = url.query["port"]
+ ports_str = url.query["port"]
elif isinstance(url.query["port"], str):
- ports = tuple(url.query["port"].split(","))
+ ports_str = tuple(url.query["port"].split(","))
+
+ ports: Optional[Tuple[Optional[int], ...]] = None
- if ports:
+ if ports_str:
try:
- ports = tuple(int(x) if x else None for x in ports)
+ ports = tuple(int(x) if x else None for x in ports_str)
except ValueError:
raise exc.ArgumentError(
- f"Received non-integer port arguments: {ports}"
+ f"Received non-integer port arguments: {ports_str}"
) from None
- if (hosts or ports) and url.host:
- raise exc.ArgumentError(
- "Can't combine fixed host and multihost URL formats"
+ if ports and (
+ (not hosts and len(ports) > 1)
+ or (
+ hosts
+ and ports
+ and len(hosts) != len(ports)
+ and (len(hosts) > 1 or len(ports) > 1)
)
-
- if ports and (not hosts or len(hosts) != len(ports)):
+ ):
raise exc.ArgumentError("number of hosts and ports don't match")
if hosts is not None:
if ports is None:
ports = tuple(None for _ in hosts)
- return hosts, ports
+ return hosts, ports # type: ignore
def do_begin_twophase(self, connection, xid):
self.do_begin(connection.connection)
"postgresql+psycopg2://scott:tiger@192.168.0.199:5432/test?sslmode=require"
)
+
Unix Domain Connections
------------------------
create_engine("postgresql+psycopg2://user:password@/dbname?host=/var/lib/postgresql")
+.. warning:: The format accepted here allows for a hostname in the main URL
+ in addition to the "host" query string argument. **When using this URL
+ format, the initial host is silently ignored**. That is, this URL::
+
+ engine = create_engine("postgresql+psycopg2://user:password@myhost1/dbname?host=myhost2")
+
+ Above, the hostname ``myhost1`` is **silently ignored and discarded.** The
+ host which is connected is the ``myhost2`` host.
+
+ This is to maintain some degree of compatibility with PostgreSQL's own URL
+ format which has been tested to behave the same way and for which tools like
+ PifPaf hardcode two hostnames.
+
.. seealso::
`PQconnectdbParams \
" for asyncpg multiple host URL",
},
),
+ (
+ # fixed host + multihost formats.
+ "postgresql+psycopg2://USER:PASS@hostfixed/DB?port=111",
+ {
+ "host": "hostfixed",
+ "port": "111",
+ "asyncpg_port": 111,
+ "dbname": "DB",
+ "user": "USER",
+ "password": "PASS",
+ },
+ ),
+ (
+ # fixed host + multihost formats. **silently ignore**
+ # the fixed host. See #10076
+ "postgresql+psycopg2://USER:PASS@hostfixed/DB?host=hostA:111",
+ {
+ "host": "hostA",
+ "port": "111",
+ "asyncpg_port": 111,
+ "dbname": "DB",
+ "user": "USER",
+ "password": "PASS",
+ },
+ ),
+ (
+ # fixed host + multihost formats. **silently ignore**
+ # the fixed host. See #10076
+ "postgresql+psycopg2://USER:PASS@hostfixed/DB"
+ "?host=hostA&port=111",
+ {
+ "host": "hostA",
+ "port": "111",
+ "asyncpg_port": 111,
+ "dbname": "DB",
+ "user": "USER",
+ "password": "PASS",
+ },
+ ),
+ (
+ # fixed host + multihost formats. **silently ignore**
+ # the fixed host. See #10076
+ "postgresql+psycopg2://USER:PASS@hostfixed/DB?host=hostA",
+ {
+ "host": "hostA",
+ "dbname": "DB",
+ "user": "USER",
+ "password": "PASS",
+ },
+ ),
+ (
+ # fixed host + multihost formats. if there is only one port
+ # or only one host after the query string, assume that's the
+ # host/port
+ "postgresql+psycopg2://USER:PASS@/DB?port=111",
+ {
+ "dbname": "DB",
+ "user": "USER",
+ "password": "PASS",
+ "port": "111",
+ "asyncpg_port": 111,
+ },
+ ),
]
for url_string, expected_psycopg in psycopg_combinations:
asyncpg_error = expected_psycopg.pop("asyncpg_error", False)
):
dialect.create_connect_args(u)
- @testing.combinations(
- ("postgresql+psycopg2://USER:PASS@hostfixed/DB?port=111",),
- ("postgresql+psycopg2://USER:PASS@hostfixed/DB?host=hostA:111",),
- (
- "postgresql+psycopg2://USER:PASS@hostfixed/DB"
- "?host=hostA&port=111",
- ),
- ("postgresql+psycopg2://USER:PASS@hostfixed/DB" "?host=hostA",),
- argnames="url_string",
- )
- @testing.combinations(
- psycopg2_dialect.dialect(),
- psycopg_dialect.dialect(),
- asyncpg_dialect.dialect(),
- argnames="dialect",
- )
- def test_dont_use_fixed_host(self, dialect, url_string):
- url_string = url_string.replace("psycopg2", dialect.driver)
-
- u = url.make_url(url_string)
- with expect_raises_message(
- exc.ArgumentError,
- "Can't combine fixed host and multihost URL formats",
- ):
- dialect.create_connect_args(u)
-
@testing.combinations(
(
"postgresql+psycopg2://USER:PASS@/DB"
"?host=hostA,hostC&port=111,222,333",
),
("postgresql+psycopg2://USER:PASS@/DB" "?host=hostA&port=111,222",),
- ("postgresql+psycopg2://USER:PASS@/DB?port=111",),
(
"postgresql+asyncpg://USER:PASS@/DB"
"?host=hostA,hostB,hostC&port=111,333",
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
