ansible-collections/google.cloud
1
0
Fork 0
mirror of https://github.com/ansible-collections/google.cloud.git synced 2025-09-29 21:13:30 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

The Union syntax for type hints is py3.10+

Browse source 
whereas the typing module Optional is supported since py3.2
This commit is contained in:
Jorge Gallegos 2025-09-02 11:39:34 -07:00
commit 361da2921f
No known key found for this signature in database
1 changed files with 27 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

54
plugins/connection/gcloud.py
View file

@ -1,4 +1,4 @@
# Copyright (c) 2025 Red Hat # Copyright (c) 202/ Red Hat
# GNU General Public License v3.0+ https://www.gnu.org/licenses/gpl-3.0.txt # GNU General Public License v3.0+ https://www.gnu.org/licenses/gpl-3.0.txt
from __future__ import absolute_import, annotations from __future__ import absolute_import, annotations
@ -221,13 +221,13 @@ DOCUMENTATION = """
D = display.Display() D = display.Display()
DEFAULT_TIMEOUT: int = 10 DEFAULT_TIMEOUT: int = 10
DEFAULT_GCLOUD: str | None = shutil.which("gcloud") DEFAULT_GCLOUD: T.Optional[str] = shutil.which("gcloud")
class Connection(connection.ConnectionBase): class Connection(connection.ConnectionBase):
"""Connections via `gcloud compute ssh`""" """Connections via `gcloud compute ssh`"""
gcloud_executable: str | None = None gcloud_executable: T.Optional[str] = None
has_pipelining = False has_pipelining = False
transport = "gcloud-ssh" # type: ignore[override] transport = "gcloud-ssh" # type: ignore[override]
@ -236,7 +236,7 @@ class Connection(connection.ConnectionBase):
super(Connection, self).__init__(*args, **kwargs) super(Connection, self).__init__(*args, **kwargs)
exec: str | None = self.get_option("gcloud_executable") exec: T.Optional[str] = self.get_option("gcloud_executable")
if exec is None: if exec is None:
self.gcloud_executable = DEFAULT_GCLOUD self.gcloud_executable = DEFAULT_GCLOUD
else: else:
@ -303,8 +303,8 @@ class Connection(connection.ConnectionBase):
self, self,
what: str, what: str,
cmd: str, cmd: str,
in_path: str | None = None, # only used for scp in_path: T.Optional[str] = None, # only used for scp
out_path: str | None = None, # only used for scp out_path: T.Optional[str] = None, # only used for scp
) -> T.List[str]: ) -> T.List[str]:
parts: T.List[str] parts: T.List[str]
@ -344,7 +344,7 @@ class Connection(connection.ConnectionBase):
def _run( def _run(
self, self,
cmd: list[str], cmd: list[str],
in_data: bytes | None, in_data: T.Optional[bytes],
sudoable: bool = True, sudoable: bool = True,
checkrc: bool = True, checkrc: bool = True,
) -> tuple[int, bytes, bytes]: ) -> tuple[int, bytes, bytes]:
@ -355,7 +355,7 @@ class Connection(connection.ConnectionBase):
# I could just not open a pty and be done with it, because I am # I could just not open a pty and be done with it, because I am
# not writing the whole pipelining and keyboard interaction just now. # not writing the whole pipelining and keyboard interaction just now.
# It may come at a later date so I am just laying the groundwork # It may come at a later date so I am just laying the groundwork
p: subprocess.Popen[bytes] | None = None p: T.Optional[subprocess.Popen[bytes]] = None
master_fd: int = 0 master_fd: int = 0
slave_fd: int = 0 slave_fd: int = 0
if in_data is None: # attempt to open a pty if in_data is None: # attempt to open a pty
@ -403,26 +403,26 @@ class Connection(connection.ConnectionBase):
return (p.returncode, stdout, stderr) return (p.returncode, stdout, stderr)
def exec_command( def exec_command(
self, cmd: str, in_data: bytes | None = None, sudoable: bool = True self, cmd: str, in_data: T.Optional[bytes] = None, sudoable: bool = True
) -> tuple[int, bytes, bytes]: ) -> tuple[int, bytes, bytes]:
"""run a command on the remote instance""" """run a command on the remote instance"""
super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable) super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
self.host: str | None = self.get_option("instance") self.host: T.Optional[str] = self.get_option("instance")
self.user: str | None = self.get_option("remote_user") self.user: T.Optional[str] = self.get_option("remote_user")
self.ssh_args: str | None = self.get_option("ssh_args") self.ssh_args: T.Optional[str] = self.get_option("ssh_args")
self.ssh_extra_args: str | None = self.get_option("ssh_extra_args") self.ssh_extra_args: T.Optional[str] = self.get_option("ssh_extra_args")
self.scp_extra_args: str | None = self.get_option("scp_extra_args") self.scp_extra_args: T.Optional[str] = self.get_option("scp_extra_args")
self.private_key_file: str | None = self.get_option("private_key_file") self.private_key_file: T.Optional[str] = self.get_option("private_key_file")
self.use_tty: bool | None = self.get_option("use_tty") self.use_tty: bool = bool(self.get_option("use_tty"))
if self.private_key_file is not None: # to silence pyright if self.private_key_file is not None: # to silence pyright
self.private_key_file = ospath.abspath( self.private_key_file = ospath.abspath(
ospath.expanduser(self.private_key_file) ospath.expanduser(self.private_key_file)
) )
self.gcp_configuration: str | None = self.get_option("configuration") self.gcp_configuration: T.Optional[str] = self.get_option("configuration")
self.gcp_project: str | None = self.get_option("project") self.gcp_project: T.Optional[str] = self.get_option("project")
self.gcp_zone: str | None = self.get_option("zone") self.gcp_zone: T.Optional[str] = self.get_option("zone")
self.timeout: int = int(self.get_option("timeout") or DEFAULT_TIMEOUT) self.timeout: int = int(self.get_option("timeout") or DEFAULT_TIMEOUT)
display.Display().vvv( display.Display().vvv(
@ -438,18 +438,18 @@ class Connection(connection.ConnectionBase):
super(Connection, self).put_file(in_path, out_path) super(Connection, self).put_file(in_path, out_path)
self.host: str | None = self.get_option("instance") self.host: T.Optional[str] = self.get_option("instance")
self.user: str | None = self.get_option("remote_user") self.user: T.Optional[str] = self.get_option("remote_user")
self.ssh_args: str | None = self.get_option("ssh_args") self.ssh_args: T.Optional[str] = self.get_option("ssh_args")
self.scp_extra_args: str | None = self.get_option("scp_extra_args") self.scp_extra_args: T.Optional[str] = self.get_option("scp_extra_args")
self.private_key_file: str | None = self.get_option("private_key_file") self.private_key_file: T.Optional[str] = self.get_option("private_key_file")
if self.private_key_file is not None: # to silence pyright if self.private_key_file is not None: # to silence pyright
self.private_key_file = ospath.abspath( self.private_key_file = ospath.abspath(
ospath.expanduser(self.private_key_file) ospath.expanduser(self.private_key_file)
) )
self.gcp_configuration: str | None = self.get_option("configuration") self.gcp_configuration: T.Optional[str] = self.get_option("configuration")
self.gcp_project: str | None = self.get_option("project") self.gcp_project: T.Optional[str] = self.get_option("project")
self.gcp_zone: str | None = self.get_option("zone") self.gcp_zone: T.Optional[str] = self.get_option("zone")
self.timeout: int = int(self.get_option("timeout") or DEFAULT_TIMEOUT) self.timeout: int = int(self.get_option("timeout") or DEFAULT_TIMEOUT)
D.vvv(f"PUT: {in_path} TO {out_path}", host=self.host) D.vvv(f"PUT: {in_path} TO {out_path}", host=self.host)