Skip to content

Commit

Permalink
Add type annotations to _ConnectionParameters and its constructing …
Browse files Browse the repository at this point in the history
…function
  • Loading branch information
DanielNoord committed Oct 23, 2024
1 parent 4686031 commit 3007b6c
Showing 1 changed file with 31 additions and 19 deletions.
50 changes: 31 additions & 19 deletions asyncpg/connect_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,20 +51,17 @@ class SSLNegotiation(compat.StrEnum):
direct = "direct"


_ConnectionParameters = collections.namedtuple(
'ConnectionParameters',
[
'user',
'password',
'database',
'ssl',
'sslmode',
'ssl_negotiation',
'server_settings',
'target_session_attrs',
'krbsrvname',
'gsslib',
])
class _ConnectionParameters(typing.NamedTuple):
user: str
password: typing.Optional[str]
database: str
ssl: typing.Union[ssl_module.SSLContext, bool, str, SSLMode, None]
sslmode: SSLMode
ssl_negotiation: SSLNegotiation
server_settings: typing.Optional[typing.Dict[str, str]]
target_session_attrs: "SessionAttribute"
krbsrvname: typing.Optional[str]
gsslib: str


_ClientConfiguration = collections.namedtuple(
Expand Down Expand Up @@ -275,10 +272,25 @@ def _dot_postgresql_path(filename) -> typing.Optional[pathlib.Path]:
return (homedir / '.postgresql' / filename).resolve()


def _parse_connect_dsn_and_args(*, dsn, host, port, user,
password, passfile, database, ssl,
direct_tls, server_settings,
target_session_attrs, krbsrvname, gsslib):
def _parse_connect_dsn_and_args(
*,
dsn: str,
host: typing.Union[str, typing.List[str], typing.Tuple[str]],
port: typing.Union[int, typing.List[int]],
user: typing.Optional[str],
password: typing.Optional[str],
passfile: typing.Union[str, pathlib.Path, None],
database: typing.Optional[str],
ssl: typing.Union[bool, None, str, SSLMode],
direct_tls: typing.Optional[bool],
server_settings: typing.Optional[typing.Dict[str, str]],
target_session_attrs: typing.Optional[str],
krbsrvname: typing.Optional[str],
gsslib: typing.Optional[str],
) -> typing.Tuple[
typing.List[typing.Union[str, typing.Tuple[str, int]]],
_ConnectionParameters,
]:
# `auth_hosts` is the version of host information for the purposes
# of reading the pgpass file.
auth_hosts = None
Expand Down Expand Up @@ -501,7 +513,7 @@ def _parse_connect_dsn_and_args(*, dsn, host, port, user,
database=database, user=user,
passfile=passfile)

addrs = []
addrs: typing.List[typing.Union[str, typing.Tuple[str, int]]] = []
have_tcp_addrs = False
for h, p in zip(host, port):
if h.startswith('/'):
Expand Down

0 comments on commit 3007b6c

Please sign in to comment.