openstack/openstacksdk
Code Issues Proposed changes

Bump Python version used for linters to 3.10

Browse Source 
This looks very large, but the only manual change is in pyproject.toml
and the bump of the ruff pre-commit hook: the rest is entirely ruff
converting our use of e.g. 'typing.Union[X, Y]' to 'X | Y', as added by
PEP-604 [1].

[1] https://peps.python.org/pep-0604/

Change-Id: I3ed176018cf78c417e751834e57412d72884a69b
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane 2025-05-07 12:33:13 +01:00
parent f19567b3cc
commit e0f57ad81f
50 changed files with 411 additions and 433 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.pre-commit-config.yaml
View File

@ -19,7 +19,7 @@ repos:
hooks:
- id: doc8
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.9.6
rev: v0.11.8
hooks:
- id: ruff
args: ['--fix', '--unsafe-fixes']

8
openstack/__init__.py
View File

@ -64,10 +64,10 @@ __all__ = [
def connect(
cloud: ty.Optional[str] = None,
app_name: ty.Optional[str] = None,
app_version: ty.Optional[str] = None,
options: ty.Optional[argparse.ArgumentParser] = None,
cloud: str | None = None,
app_name: str | None = None,
app_version: str | None = None,
options: argparse.ArgumentParser | None = None,
load_yaml_config: bool = True,
load_envvars: bool = True,
**kwargs: ty.Any,

10
openstack/_log.py
View File

@ -19,8 +19,8 @@ import typing as ty
def setup_logging(
name: str,
handlers: ty.Optional[list[logging.Handler]] = None,
level: ty.Optional[int] = None,
handlers: list[logging.Handler] | None = None,
level: int | None = None,
) -> logging.Logger:
"""Set up logging for a named logger.
@ -50,11 +50,11 @@ def setup_logging(
def enable_logging(
debug: bool = False,
http_debug: bool = False,
path: ty.Optional[str] = None,
stream: ty.Optional[ty.TextIO] = None,
path: str | None = None,
stream: ty.TextIO | None = None,
format_stream: bool = False,
format_template: str = '%(asctime)s %(levelname)s: %(name)s %(message)s',
handlers: ty.Optional[list[logging.Handler]] = None,
handlers: list[logging.Handler] | None = None,
) -> None:
"""Enable logging output.

10
openstack/accelerator/v2/_proxy.py
View File

@ -260,11 +260,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -300,7 +300,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

3
openstack/baremetal/configdrive.py
View File

@ -20,7 +20,6 @@ import os
import shutil
import subprocess
import tempfile
import typing as ty
@contextlib.contextmanager
@ -110,7 +109,7 @@ def pack(path: str) -> str:
with tempfile.NamedTemporaryFile() as tmpfile:
# NOTE(toabctl): Luckily, genisoimage, mkisofs and xorrisofs understand
# the same parameters which are currently used.
error: ty.Optional[Exception]
error: Exception | None
for c in ['genisoimage', 'mkisofs', 'xorrisofs']:
try:
p = subprocess.Popen( # noqa: S603

18
openstack/baremetal/v1/_proxy.py
View File

@ -241,7 +241,7 @@ class Proxy(proxy.Proxy):
def call_driver_vendor_passthru(
self,
driver: ty.Union[str, _driver.Driver],
driver: str | _driver.Driver,
verb: str,
method: str,
body: object = None,
@ -1171,12 +1171,12 @@ class Proxy(proxy.Proxy):
def attach_vif_to_node(
self,
node: ty.Union[_node.Node, str],
node: _node.Node | str,
vif_id: str,
retry_on_conflict: bool = True,
*,
port_id: ty.Optional[str] = None,
port_group_id: ty.Optional[str] = None,
port_id: str | None = None,
port_group_id: str | None = None,
) -> None:
"""Attach a VIF to the node.
@ -1868,11 +1868,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -1908,7 +1908,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

3
openstack/baremetal/v1/driver.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import typing as ty
from keystoneauth1 import adapter
import requests
@ -162,7 +161,7 @@ class Driver(resource.Resource):
session: adapter.Adapter,
verb: str,
method: str,
body: ty.Optional[dict] = None,
body: dict | None = None,
) -> requests.Response:
"""Call a vendor specific passthru method

5
openstack/baremetal/v1/node.py
View File

@ -12,7 +12,6 @@
import collections
import enum
import typing as ty
import warnings
from keystoneauth1 import adapter
@ -912,8 +911,8 @@ class Node(_common.Resource):
vif_id: str,
retry_on_conflict: bool = True,
*,
port_id: ty.Optional[str] = None,
port_group_id: ty.Optional[str] = None,
port_id: str | None = None,
port_group_id: str | None = None,
) -> None:
"""Attach a VIF to the node.

10
openstack/baremetal_introspection/v1/_proxy.py
View File

@ -241,11 +241,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -281,7 +281,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

30
openstack/block_storage/v2/_proxy.py
View File

@ -983,7 +983,7 @@ class Proxy(proxy.Proxy):
name_or_id: str,
ignore_missing: ty.Literal[True] = True,
**query: ty.Any,
) -> ty.Optional[_service.Service]: ...
) -> _service.Service | None: ...
@ty.overload
def find_service(
@ -1001,14 +1001,14 @@ class Proxy(proxy.Proxy):
name_or_id: str,
ignore_missing: bool,
**query: ty.Any,
) -> ty.Optional[_service.Service]: ...
) -> _service.Service | None: ...
def find_service(
self,
name_or_id: str,
ignore_missing: bool = True,
**query: ty.Any,
) -> ty.Optional[_service.Service]:
) -> _service.Service | None:
"""Find a single service
:param name_or_id: The name or ID of a service
@ -1047,7 +1047,7 @@ class Proxy(proxy.Proxy):
def enable_service(
self,
service: ty.Union[str, _service.Service],
service: str | _service.Service,
) -> _service.Service:
"""Enable a service
@ -1062,9 +1062,9 @@ class Proxy(proxy.Proxy):
def disable_service(
self,
service: ty.Union[str, _service.Service],
service: str | _service.Service,
*,
reason: ty.Optional[str] = None,
reason: str | None = None,
) -> _service.Service:
"""Disable a service
@ -1080,7 +1080,7 @@ class Proxy(proxy.Proxy):
def thaw_service(
self,
service: ty.Union[str, _service.Service],
service: str | _service.Service,
) -> _service.Service:
"""Thaw a service
@ -1095,7 +1095,7 @@ class Proxy(proxy.Proxy):
def freeze_service(
self,
service: ty.Union[str, _service.Service],
service: str | _service.Service,
) -> _service.Service:
"""Freeze a service
@ -1110,9 +1110,9 @@ class Proxy(proxy.Proxy):
def failover_service(
self,
service: ty.Union[str, _service.Service],
service: str | _service.Service,
*,
backend_id: ty.Optional[str] = None,
backend_id: str | None = None,
) -> _service.Service:
"""Failover a service
@ -1332,11 +1332,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str = 'available',
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -1375,7 +1375,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

44
openstack/block_storage/v3/_proxy.py
View File

@ -2088,7 +2088,7 @@ class Proxy(proxy.Proxy):
name_or_id: str,
ignore_missing: ty.Literal[True] = True,
**query: ty.Any,
) -> ty.Optional[_service.Service]: ...
) -> _service.Service | None: ...
@ty.overload
def find_service(
@ -2106,14 +2106,14 @@ class Proxy(proxy.Proxy):
name_or_id: str,
ignore_missing: bool,
**query: ty.Any,
) -> ty.Optional[_service.Service]: ...
) -> _service.Service | None: ...
def find_service(
self,
name_or_id: str,
ignore_missing: bool = True,
**query: ty.Any,
) -> ty.Optional[_service.Service]:
) -> _service.Service | None:
"""Find a single service
:param name_or_id: The name or ID of a service
@ -2152,7 +2152,7 @@ class Proxy(proxy.Proxy):
def enable_service(
self,
service: ty.Union[str, _service.Service],
service: str | _service.Service,
) -> _service.Service:
"""Enable a service
@ -2167,9 +2167,9 @@ class Proxy(proxy.Proxy):
def disable_service(
self,
service: ty.Union[str, _service.Service],
service: str | _service.Service,
*,
reason: ty.Optional[str] = None,
reason: str | None = None,
) -> _service.Service:
"""Disable a service
@ -2185,7 +2185,7 @@ class Proxy(proxy.Proxy):
def thaw_service(
self,
service: ty.Union[str, _service.Service],
service: str | _service.Service,
) -> _service.Service:
"""Thaw a service
@ -2200,7 +2200,7 @@ class Proxy(proxy.Proxy):
def freeze_service(
self,
service: ty.Union[str, _service.Service],
service: str | _service.Service,
) -> _service.Service:
"""Freeze a service
@ -2217,9 +2217,9 @@ class Proxy(proxy.Proxy):
self,
*,
level: _service.Level,
binary: ty.Optional[_service.Binary] = None,
server: ty.Optional[str] = None,
prefix: ty.Optional[str] = None,
binary: _service.Binary | None = None,
server: str | None = None,
prefix: str | None = None,
) -> None:
"""Set log level for services.
@ -2239,9 +2239,9 @@ class Proxy(proxy.Proxy):
def get_service_log_levels(
self,
*,
binary: ty.Optional[_service.Binary] = None,
server: ty.Optional[str] = None,
prefix: ty.Optional[str] = None,
binary: _service.Binary | None = None,
server: str | None = None,
prefix: str | None = None,
) -> ty.Generator[_service.LogLevel, None, None]:
"""Get log level for services.
@ -2259,10 +2259,10 @@ class Proxy(proxy.Proxy):
def failover_service(
self,
service: ty.Union[str, _service.Service],
service: str | _service.Service,
*,
cluster: ty.Optional[str] = None,
backend_id: ty.Optional[str] = None,
cluster: str | None = None,
backend_id: str | None = None,
) -> _service.Service:
"""Failover a service
@ -2404,11 +2404,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str = 'available',
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -2447,7 +2447,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

12
openstack/block_storage/v3/service.py
View File

@ -175,9 +175,9 @@ class Service(resource.Resource):
session: ksa_adapter.Adapter,
*,
level: Level,
binary: ty.Optional[Binary] = None,
server: ty.Optional[str] = None,
prefix: ty.Optional[str] = None,
binary: Binary | None = None,
server: str | None = None,
prefix: str | None = None,
) -> None:
"""Set log level for services.
@ -209,9 +209,9 @@ class Service(resource.Resource):
cls,
session: ksa_adapter.Adapter,
*,
binary: ty.Optional[Binary] = None,
server: ty.Optional[str] = None,
prefix: ty.Optional[str] = None,
binary: Binary | None = None,
server: str | None = None,
prefix: str | None = None,
) -> ty.Generator[LogLevel, None, None]:
"""Get log level for services.

4
openstack/cloud/_compute.py
View File

@ -278,7 +278,7 @@ class ComputeCloudMixin(_network_common.NetworkCommonCloudMixin):
self.log.debug('Server %s not found', server)
return None, None
if not isinstance(security_groups, (list, tuple)):
if not isinstance(security_groups, list | tuple):
security_groups = [security_groups]
sec_group_objs = []
@ -1812,7 +1812,7 @@ class ComputeCloudMixin(_network_common.NetworkCommonCloudMixin):
:raises: :class:`~openstack.exceptions.SDKException` on operation
error.
"""
if isinstance(name_or_id, (str, bytes)) and not name_or_id.isdigit():
if isinstance(name_or_id, str | bytes) and not name_or_id.isdigit():
aggregate = self.get_aggregate(name_or_id)
if not aggregate:
self.log.debug(

19
openstack/cloud/_image.py
View File

@ -31,21 +31,20 @@ if ty.TYPE_CHECKING:
class ImageCloudMixin(openstackcloud._OpenStackCloudMixin):
def __init__(
self,
cloud: ty.Optional[str] = None,
cloud: str | None = None,
config: ty.Optional['cloud_region.CloudRegion'] = None,
session: ty.Optional['ks_session.Session'] = None,
app_name: ty.Optional[str] = None,
app_version: ty.Optional[str] = None,
extra_services: ty.Optional[
list['service_description.ServiceDescription']
] = None,
app_name: str | None = None,
app_version: str | None = None,
extra_services: list['service_description.ServiceDescription']
| None = None,
strict: bool = False,
use_direct_get: ty.Optional[bool] = None,
use_direct_get: bool | None = None,
task_manager: ty.Any = None,
rate_limit: ty.Union[float, dict[str, float], None] = None,
rate_limit: float | dict[str, float] | None = None,
oslo_conf: ty.Optional['cfg.ConfigOpts'] = None,
service_types: ty.Optional[list[str]] = None,
global_request_id: ty.Optional[str] = None,
service_types: list[str] | None = None,
global_request_id: str | None = None,
strict_proxies: bool = False,
pool_executor: ty.Optional['concurrent.futures.Executor'] = None,
**kwargs: ty.Any,

29
openstack/cloud/openstackcloud.py
View File

@ -72,23 +72,22 @@ class _OpenStackCloudMixin(_services_mixin.ServicesMixin):
def __init__(
self,
cloud: ty.Optional[str] = None,
config: ty.Optional[cloud_region.CloudRegion] = None,
cloud: str | None = None,
config: cloud_region.CloudRegion | None = None,
session: ty.Optional['ks_session.Session'] = None,
app_name: ty.Optional[str] = None,
app_version: ty.Optional[str] = None,
extra_services: ty.Optional[
list['service_description.ServiceDescription']
] = None,
app_name: str | None = None,
app_version: str | None = None,
extra_services: list['service_description.ServiceDescription']
| None = None,
strict: bool = False,
use_direct_get: ty.Optional[bool] = None,
use_direct_get: bool | None = None,
task_manager: ty.Any = None,
rate_limit: ty.Union[float, dict[str, float], None] = None,
rate_limit: float | dict[str, float] | None = None,
oslo_conf: ty.Optional['cfg.ConfigOpts'] = None,
service_types: ty.Optional[list[str]] = None,
global_request_id: ty.Optional[str] = None,
service_types: list[str] | None = None,
global_request_id: str | None = None,
strict_proxies: bool = False,
pool_executor: ty.Optional[concurrent.futures.Executor] = None,
pool_executor: concurrent.futures.Executor | None = None,
**kwargs: ty.Any,
):
"""Create a connection to a cloud.
@ -300,9 +299,9 @@ class _OpenStackCloudMixin(_services_mixin.ServicesMixin):
def __exit__(
self,
exc_type: ty.Optional[type[BaseException]],
exc_value: ty.Optional[BaseException],
traceback: ty.Optional[types.TracebackType],
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: types.TracebackType | None,
) -> None:
self.close()

10
openstack/clustering/v1/_proxy.py
View File

@ -1077,11 +1077,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -1117,7 +1117,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

18
openstack/compute/v2/_proxy.py
View File

@ -2648,10 +2648,10 @@ class Proxy(proxy.Proxy):
self,
server: _server.Server,
status: str = 'ACTIVE',
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = 120,
callback: ty.Callable[[int], None] | None = None,
) -> _server.Server:
"""Wait for a server to be in a particular status.
@ -2694,11 +2694,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -2734,7 +2734,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

8
openstack/config/__init__.py
View File

@ -24,10 +24,10 @@ if ty.TYPE_CHECKING:
# TODO(stephenfin): Expand kwargs once we've typed OpenstackConfig.get_one
def get_cloud_region(
service_key: ty.Optional[str] = None,
options: ty.Optional[argparse.ArgumentParser] = None,
app_name: ty.Optional[str] = None,
app_version: ty.Optional[str] = None,
service_key: str | None = None,
options: argparse.ArgumentParser | None = None,
app_name: str | None = None,
app_version: str | None = None,
load_yaml_config: bool = True,
load_envvars: bool = True,
**kwargs: ty.Any,

58
openstack/config/cloud_region.py
View File

@ -74,7 +74,7 @@ _ENOENT = object()
class _PasswordCallback(ty.Protocol):
def __call__(self, prompt: ty.Optional[str] = None) -> str: ...
def __call__(self, prompt: str | None = None) -> str: ...
def _make_key(key, service_type):
@ -108,11 +108,11 @@ def _get_implied_microversion(version):
def from_session(
session: ks_session.Session,
name: ty.Optional[str] = None,
region_name: ty.Optional[str] = None,
name: str | None = None,
region_name: str | None = None,
force_ipv4: bool = False,
app_name: ty.Optional[str] = None,
app_version: ty.Optional[str] = None,
app_name: str | None = None,
app_version: str | None = None,
**kwargs: ty.Any,
) -> 'CloudRegion':
"""Construct a CloudRegion from an existing `keystoneauth1.session.Session`
@ -152,8 +152,8 @@ def from_session(
def from_conf(
conf: 'cfg.ConfigOpts',
session: ty.Optional[ks_session.Session] = None,
service_types: ty.Optional[list[str]] = None,
session: ks_session.Session | None = None,
service_types: list[str] | None = None,
**kwargs: ty.Any,
) -> 'CloudRegion':
"""Create a CloudRegion from oslo.config ConfigOpts.
@ -291,29 +291,29 @@ class CloudRegion:
def __init__(
self,
name: ty.Optional[str] = None,
region_name: ty.Optional[str] = None,
config: ty.Optional[dict[str, ty.Any]] = None,
name: str | None = None,
region_name: str | None = None,
config: dict[str, ty.Any] | None = None,
force_ipv4: bool = False,
auth_plugin: ty.Optional[plugin.BaseAuthPlugin] = None,
auth_plugin: plugin.BaseAuthPlugin | None = None,
openstack_config: ty.Optional['loader.OpenStackConfig'] = None,
session_constructor: ty.Optional[type[ks_session.Session]] = None,
app_name: ty.Optional[str] = None,
app_version: ty.Optional[str] = None,
session: ty.Optional[ks_session.Session] = None,
discovery_cache: ty.Optional[dict[str, discover.Discover]] = None,
extra_config: ty.Optional[dict[str, ty.Any]] = None,
session_constructor: type[ks_session.Session] | None = None,
app_name: str | None = None,
app_version: str | None = None,
session: ks_session.Session | None = None,
discovery_cache: dict[str, discover.Discover] | None = None,
extra_config: dict[str, ty.Any] | None = None,
cache_expiration_time: int = 0,
cache_expirations: ty.Optional[dict[str, int]] = None,
cache_path: ty.Optional[str] = None,
cache_expirations: dict[str, int] | None = None,
cache_path: str | None = None,
cache_class: str = 'dogpile.cache.null',
cache_arguments: ty.Optional[dict[str, ty.Any]] = None,
password_callback: ty.Optional[_PasswordCallback] = None,
statsd_host: ty.Optional[str] = None,
statsd_port: ty.Optional[str] = None,
statsd_prefix: ty.Optional[str] = None,
cache_arguments: dict[str, ty.Any] | None = None,
password_callback: _PasswordCallback | None = None,
statsd_host: str | None = None,
statsd_port: str | None = None,
statsd_prefix: str | None = None,
# TODO(stephenfin): Add better types
influxdb_config: ty.Optional[dict[str, ty.Any]] = None,
influxdb_config: dict[str, ty.Any] | None = None,
collector_registry: ty.Optional[
'prometheus_client.CollectorRegistry'
] = None,
@ -577,7 +577,7 @@ class CloudRegion:
def get_service_name(self, service_type):
return self._get_config('service_name', service_type)
def get_endpoint(self, service_type: str) -> ty.Optional[str]:
def get_endpoint(self, service_type: str) -> str | None:
auth = self.config.get('auth', {})
value = self._get_config('endpoint_override', service_type)
if not value:
@ -618,9 +618,9 @@ class CloudRegion:
def get_endpoint_from_catalog(
self,
service_type: str,
interface: ty.Optional[str] = None,
region_name: ty.Optional[str] = None,
) -> ty.Optional[str]:
interface: str | None = None,
region_name: str | None = None,
) -> str | None:
"""Return the endpoint for a given service as found in the catalog.
For values respecting endpoint overrides, see

32
openstack/config/loader.py
View File

@ -158,22 +158,22 @@ class OpenStackConfig:
def __init__(
self,
config_files: ty.Optional[list[str]] = None,
vendor_files: ty.Optional[list[str]] = None,
override_defaults: ty.Optional[dict[str, ty.Any]] = None,
force_ipv4: ty.Optional[bool] = None,
envvar_prefix: ty.Optional[str] = None,
secure_files: ty.Optional[list[str]] = None,
pw_func: ty.Optional[cloud_region._PasswordCallback] = None,
session_constructor: ty.Optional[type[session.Session]] = None,
app_name: ty.Optional[str] = None,
app_version: ty.Optional[str] = None,
config_files: list[str] | None = None,
vendor_files: list[str] | None = None,
override_defaults: dict[str, ty.Any] | None = None,
force_ipv4: bool | None = None,
envvar_prefix: str | None = None,
secure_files: list[str] | None = None,
pw_func: cloud_region._PasswordCallback | None = None,
session_constructor: type[session.Session] | None = None,
app_name: str | None = None,
app_version: str | None = None,
load_yaml_config: bool = True,
load_envvars: bool = True,
statsd_host: ty.Optional[str] = None,
statsd_port: ty.Optional[str] = None,
statsd_prefix: ty.Optional[str] = None,
influxdb_config: ty.Optional[dict[str, ty.Any]] = None,
statsd_host: str | None = None,
statsd_port: str | None = None,
statsd_prefix: str | None = None,
influxdb_config: dict[str, ty.Any] | None = None,
):
self.log = _log.setup_logging('openstack.config')
self._session_constructor = session_constructor
@ -1221,9 +1221,9 @@ class OpenStackConfig:
def get_one(
self,
cloud: ty.Optional[str] = None,
cloud: str | None = None,
validate: bool = True,
argparse: ty.Optional[argparse_mod.Namespace] = None,
argparse: argparse_mod.Namespace | None = None,
**kwargs: ty.Any,
) -> cloud_region.CloudRegion:
"""Retrieve a single CloudRegion and merge additional options

37
openstack/connection.py
View File

@ -321,9 +321,9 @@ _logger = _log.setup_logging('openstack')
def from_config(
cloud: ty.Optional[str] = None,
cloud: str | None = None,
config: ty.Optional['cloud_region.CloudRegion'] = None,
options: ty.Optional[argparse.Namespace] = None,
options: argparse.Namespace | None = None,
**kwargs: ty.Any,
) -> 'Connection':
"""Create a Connection using openstack.config
@ -371,23 +371,22 @@ class Connection(
):
def __init__(
self,
cloud: ty.Optional[str] = None,
cloud: str | None = None,
config: ty.Optional['cloud_region.CloudRegion'] = None,
session: ty.Optional[ks_session.Session] = None,
app_name: ty.Optional[str] = None,
app_version: ty.Optional[str] = None,
extra_services: ty.Optional[
list[service_description.ServiceDescription]
] = None,
session: ks_session.Session | None = None,
app_name: str | None = None,
app_version: str | None = None,
extra_services: list[service_description.ServiceDescription]
| None = None,
strict: bool = False,
use_direct_get: ty.Optional[bool] = None,
use_direct_get: bool | None = None,
task_manager: ty.Any = None,
rate_limit: ty.Union[float, dict[str, float], None] = None,
rate_limit: float | dict[str, float] | None = None,
oslo_conf: ty.Optional['cfg.ConfigOpts'] = None,
service_types: ty.Optional[list[str]] = None,
global_request_id: ty.Optional[str] = None,
service_types: list[str] | None = None,
global_request_id: str | None = None,
strict_proxies: bool = False,
pool_executor: ty.Optional[concurrent.futures.Executor] = None,
pool_executor: concurrent.futures.Executor | None = None,
**kwargs: ty.Any,
):
"""Create a connection to a cloud.
@ -623,8 +622,8 @@ class Connection(
# Utility function to help with the stripping below.
def pop_keys(
params: dict[str, dict[str, ty.Optional[str]]],
auth: dict[str, ty.Optional[str]],
params: dict[str, dict[str, str | None]],
auth: dict[str, str | None],
name_key: str,
id_key: str,
) -> None:
@ -695,9 +694,9 @@ class Connection(
def endpoint_for(
self,
service_type: str,
interface: ty.Optional[str] = None,
region_name: ty.Optional[str] = None,
) -> ty.Optional[str]:
interface: str | None = None,
region_name: str | None = None,
) -> str | None:
"""Return the endpoint for a given service.
Respects config values for Connection, including

10
openstack/container_infrastructure_management/v1/_proxy.py
View File

@ -266,11 +266,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -306,7 +306,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

10
openstack/database/v1/_proxy.py
View File

@ -342,11 +342,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -382,7 +382,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

5
openstack/dns/v2/_base.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import typing as ty
import urllib.parse
from openstack import exceptions
@ -80,7 +79,7 @@ class Resource(resource.Resource):
all_projects=None,
**params,
):
headers: ty.Union[ty.Union[dict[str, str], None]] = (
headers: dict[str, str] | None = (
{} if project_id or all_projects else None
)
@ -95,7 +94,7 @@ class Resource(resource.Resource):
@classmethod
def _get_next_link(cls, uri, response, data, marker, limit, total_yielded):
next_link = None
params: dict[str, ty.Union[list[str], str]] = {}
params: dict[str, list[str] | str] = {}
if isinstance(data, dict):
links = data.get('links')
if links:

10
openstack/dns/v2/_proxy.py
View File

@ -720,11 +720,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -760,7 +760,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

26
openstack/exceptions.py
View File

@ -33,9 +33,7 @@ if ty.TYPE_CHECKING:
class SDKException(Exception):
"""The base exception class for all exceptions this library raises."""
def __init__(
self, message: ty.Optional[str] = None, extra_data: ty.Any = None
):
def __init__(self, message: str | None = None, extra_data: ty.Any = None):
self.message = self.__class__.__name__ if message is None else message
self.extra_data = extra_data
super().__init__(self.message)
@ -44,21 +42,21 @@ class SDKException(Exception):
class EndpointNotFound(SDKException):
"""A mismatch occurred between what the client and server expect."""
def __init__(self, message: ty.Optional[str] = None):
def __init__(self, message: str | None = None):
super().__init__(message)
class InvalidResponse(SDKException):
"""The response from the server is not valid for this request."""
def __init__(self, message: ty.Optional[str] = None):
def __init__(self, message: str | None = None):
super().__init__(message)
class InvalidRequest(SDKException):
"""The request to the server is not valid."""
def __init__(self, message: ty.Optional[str] = None):
def __init__(self, message: str | None = None):
super().__init__(message)
@ -66,15 +64,15 @@ class HttpException(SDKException, _rex.HTTPError):
"""The base exception for all HTTP error responses."""
source: str
status_code: ty.Optional[int]
status_code: int | None
def __init__(
self,
message: ty.Optional[str] = 'Error',
response: ty.Optional[requests.Response] = None,
http_status: ty.Optional[int] = None,
details: ty.Optional[str] = None,
request_id: ty.Optional[str] = None,
message: str | None = 'Error',
response: requests.Response | None = None,
http_status: int | None = None,
details: str | None = None,
request_id: str | None = None,
):
if http_status is not None:
warnings.warn(
@ -190,7 +188,7 @@ class InvalidResourceQuery(SDKException):
"""Invalid query params for resource."""
def _extract_message(obj: ty.Any) -> ty.Optional[str]:
def _extract_message(obj: ty.Any) -> str | None:
if isinstance(obj, dict):
# Most of services: compute, network
if obj.get('message'):
@ -212,7 +210,7 @@ def _extract_message(obj: ty.Any) -> ty.Optional[str]:
def raise_from_response(
response: requests.Response,
error_message: ty.Optional[str] = None,
error_message: str | None = None,
) -> None:
"""Raise an instance of an HTTPException based on keystoneauth response."""
if response.status_code < 400:

48
openstack/fields.py
View File

@ -84,25 +84,19 @@ def _convert_type(
def _convert_type(
value: _T1,
data_type: ty.Optional[
type[
ty.Union[
_T3,
list[ty.Any],
dict[ty.Any, ty.Any],
format.Formatter[_T2],
],
]
],
list_type: ty.Optional[type[_T3]] = None,
) -> ty.Union[_T1, _T3, list[_T3], list[_T1], dict[ty.Any, ty.Any], _T2]:
data_type: type[
_T3 | list[ty.Any] | dict[ty.Any, ty.Any] | format.Formatter[_T2],
]
| None,
list_type: type[_T3] | None = None,
) -> _T1 | _T3 | list[_T3] | list[_T1] | dict[ty.Any, ty.Any] | _T2:
# This should allow handling list of dicts that have their own
# Component type directly. See openstack/compute/v2/limits.py
# and the RateLimit type for an example.
if data_type is None:
return value
elif issubclass(data_type, list):
if isinstance(value, (list, set, tuple)):
if isinstance(value, list | set | tuple):
if not list_type:
return data_type(value)
return [_convert_type(x, list_type) for x in value]
@ -121,8 +115,8 @@ def _convert_type(
return data_type.deserialize(value)
elif issubclass(data_type, bool):
return data_type(value)
elif issubclass(data_type, (int, float)):
if isinstance(value, (int, float)):
elif issubclass(data_type, int | float):
if isinstance(value, int | float):
return data_type(value)
if isinstance(value, str):
if issubclass(data_type, int) and value.isdigit():
@ -149,28 +143,28 @@ class _BaseComponent(abc.ABC):
_map_cls: ty.ClassVar[type[ty.MutableMapping[str, ty.Any]]] = dict
name: str
data_type: ty.Optional[ty.Any]
data_type: ty.Any | None
default: ty.Any
alias: ty.Optional[str]
aka: ty.Optional[str]
alias: str | None
aka: str | None
alternate_id: bool
list_type: ty.Optional[ty.Any]
list_type: ty.Any | None
coerce_to_default: bool
deprecated: bool
deprecation_reason: ty.Optional[str]
deprecation_reason: str | None
def __init__(
self,
name: str,
type: ty.Optional[ty.Any] = None,
type: ty.Any | None = None,
default: ty.Any = None,
alias: ty.Optional[str] = None,
aka: ty.Optional[str] = None,
alias: str | None = None,
aka: str | None = None,
alternate_id: bool = False,
list_type: ty.Optional[ty.Any] = None,
list_type: ty.Any | None = None,
coerce_to_default: bool = False,
deprecated: bool = False,
deprecation_reason: ty.Optional[str] = None,
deprecation_reason: str | None = None,
):
"""A typed descriptor for a component that makes up a Resource
@ -214,7 +208,7 @@ class _BaseComponent(abc.ABC):
def __get__(
self,
instance: object,
owner: ty.Optional[type[object]] = None,
owner: type[object] | None = None,
) -> ty.Any:
if instance is None:
return self
@ -261,7 +255,7 @@ class _BaseComponent(abc.ABC):
return _convert_type(value, self.data_type, self.list_type)
@property
def type(self) -> ty.Optional[ty.Any]:
def type(self) -> ty.Any | None:
# deprecated alias proxy
return self.data_type

10
openstack/identity/v2/_proxy.py
View File

@ -282,11 +282,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -322,7 +322,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

10
openstack/identity/v3/_proxy.py
View File

@ -2396,11 +2396,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -2436,7 +2436,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

10
openstack/image/v1/_proxy.py
View File

@ -482,11 +482,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -522,7 +522,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

10
openstack/image/v2/_proxy.py
View File

@ -1900,11 +1900,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -1940,7 +1940,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

10
openstack/instance_ha/v1/_proxy.py
View File

@ -267,11 +267,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -307,7 +307,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

10
openstack/key_manager/v1/_proxy.py
View File

@ -276,11 +276,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -316,7 +316,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

10
openstack/load_balancer/v2/_proxy.py
View File

@ -1292,11 +1292,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -1332,7 +1332,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

10
openstack/message/v2/_proxy.py
View File

@ -311,11 +311,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -351,7 +351,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

3
openstack/message/v2/message.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import typing as ty
import uuid
from openstack import resource
@ -56,7 +55,7 @@ class Message(resource.Resource):
# FIXME(stephenfin): This is actually a query arg but we need it for
# deletions and resource.delete doesn't respect these currently
claim_id: ty.Optional[str] = None
claim_id: str | None = None
def post(self, session, messages):
request = self._prepare_request(requires_id=False, prepend_key=True)

28
openstack/network/v2/_proxy.py
View File

@ -199,9 +199,9 @@ class Proxy(proxy.Proxy):
def _update(
self,
resource_type: type[resource.ResourceT],
value: ty.Union[str, resource.ResourceT, None],
base_path: ty.Optional[str] = None,
if_revision: ty.Optional[int] = None,
value: str | resource.ResourceT | None,
base_path: str | None = None,
if_revision: int | None = None,
**attrs: ty.Any,
) -> resource.ResourceT:
if (
@ -217,11 +217,11 @@ class Proxy(proxy.Proxy):
def _delete(
self,
resource_type: type[resource.ResourceT],
value: ty.Union[str, resource.ResourceT, None],
value: str | resource.ResourceT | None,
ignore_missing: bool = True,
if_revision: ty.Optional[int] = None,
if_revision: int | None = None,
**attrs: ty.Any,
) -> ty.Optional[resource.ResourceT]:
) -> resource.ResourceT | None:
if (
issubclass(resource_type, _base.NetworkResource)
and if_revision is not None
@ -322,7 +322,7 @@ class Proxy(proxy.Proxy):
def update_address_group(
self,
address_group: ty.Union[str, _address_group.AddressGroup],
address_group: str | _address_group.AddressGroup,
**attrs: ty.Any,
) -> _address_group.AddressGroup:
"""Update an address group
@ -2998,8 +2998,8 @@ class Proxy(proxy.Proxy):
def update_port(
self,
port: ty.Union[str, _port.Port],
if_revision: ty.Optional[int] = None,
port: str | _port.Port,
if_revision: int | None = None,
**attrs: ty.Any,
) -> _port.Port:
"""Update a port
@ -7238,11 +7238,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -7278,7 +7278,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

12
openstack/object_store/v1/_proxy.py
View File

@ -1044,7 +1044,7 @@ class Proxy(proxy.Proxy):
method.upper(),
)
expiration: ty.Union[float, int]
expiration: float | int
if not absolute:
expiration = _get_expiration(timestamp)
else:
@ -1129,11 +1129,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -1169,7 +1169,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

2
openstack/orchestration/util/template_utils.py
View File

@ -89,7 +89,7 @@ def resolve_template_get_files(
return False
def recurse_if(value):
return isinstance(value, (dict, list))
return isinstance(value, dict | list)
get_file_contents(
template,

10
openstack/orchestration/v1/_proxy.py
View File

@ -577,11 +577,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -617,7 +617,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

10
openstack/placement/v1/_proxy.py
View File

@ -469,11 +469,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -509,7 +509,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

154
openstack/resource.py
View File

@ -65,15 +65,15 @@ ResourceT = ty.TypeVar('ResourceT', bound='Resource')
# with Any rather than generating super complex types
def Body(
name: str,
type: ty.Optional[ty.Any] = None,
type: ty.Any | None = None,
default: ty.Any = None,
alias: ty.Optional[str] = None,
aka: ty.Optional[str] = None,
alias: str | None = None,
aka: str | None = None,
alternate_id: bool = False,
list_type: ty.Optional[ty.Any] = None,
list_type: ty.Any | None = None,
coerce_to_default: bool = False,
deprecated: bool = False,
deprecation_reason: ty.Optional[str] = None,
deprecation_reason: str | None = None,
) -> ty.Any:
return fields.Body(
name,
@ -91,15 +91,15 @@ def Body(
def Header(
name: str,
type: ty.Optional[ty.Any] = None,
type: ty.Any | None = None,
default: ty.Any = None,
alias: ty.Optional[str] = None,
aka: ty.Optional[str] = None,
alias: str | None = None,
aka: str | None = None,
alternate_id: bool = False,
list_type: ty.Optional[ty.Any] = None,
list_type: ty.Any | None = None,
coerce_to_default: bool = False,
deprecated: bool = False,
deprecation_reason: ty.Optional[str] = None,
deprecation_reason: str | None = None,
) -> ty.Any:
return fields.Header(
name,
@ -117,15 +117,15 @@ def Header(
def URI(
name: str,
type: ty.Optional[ty.Any] = None,
type: ty.Any | None = None,
default: ty.Any = None,
alias: ty.Optional[str] = None,
aka: ty.Optional[str] = None,
alias: str | None = None,
aka: str | None = None,
alternate_id: bool = False,
list_type: ty.Optional[ty.Any] = None,
list_type: ty.Any | None = None,
coerce_to_default: bool = False,
deprecated: bool = False,
deprecation_reason: ty.Optional[str] = None,
deprecation_reason: str | None = None,
) -> ty.Any:
return fields.URI(
name,
@ -143,15 +143,15 @@ def URI(
def Computed(
name: str,
type: ty.Optional[ty.Any] = None,
type: ty.Any | None = None,
default: ty.Any = None,
alias: ty.Optional[str] = None,
aka: ty.Optional[str] = None,
alias: str | None = None,
aka: str | None = None,
alternate_id: bool = False,
list_type: ty.Optional[ty.Any] = None,
list_type: ty.Any | None = None,
coerce_to_default: bool = False,
deprecated: bool = False,
deprecation_reason: ty.Optional[str] = None,
deprecation_reason: str | None = None,
) -> ty.Any:
return fields.Computed(
name,
@ -251,7 +251,7 @@ class QueryParameters:
parameters, ``limit`` and ``marker``. These are the most common
query parameters used for listing resources in OpenStack APIs.
"""
self._mapping: dict[str, ty.Union[str, dict]] = {}
self._mapping: dict[str, str | dict] = {}
if include_pagination_defaults:
self._mapping.update({"limit": "limit", "marker": "marker"})
self._mapping.update({name: name for name in names})
@ -351,11 +351,11 @@ class Resource(dict):
# will work properly.
#: Singular form of key for resource.
resource_key: ty.Optional[str] = None
resource_key: str | None = None
#: Plural form of key for resource.
resources_key: ty.Optional[str] = None
resources_key: str | None = None
#: Key used for pagination links
pagination_key: ty.Optional[str] = None
pagination_key: str | None = None
#: The ID of this resource.
id: str = Body("id")
@ -399,16 +399,16 @@ class Resource(dict):
#: Do calls for this resource require an id
requires_id = True
#: Whether create requires an ID (determined from method if None).
create_requires_id: ty.Optional[bool] = None
create_requires_id: bool | None = None
#: Whether create should exclude ID in the body of the request.
create_exclude_id_from_body = False
#: Do responses for this resource have bodies
has_body = True
#: Does create returns a body (if False requires ID), defaults to has_body
create_returns_body: ty.Optional[bool] = None
create_returns_body: bool | None = None
#: Maximum microversion to use for getting/creating/updating the Resource
_max_microversion: ty.Optional[str] = None
_max_microversion: str | None = None
#: API microversion (string or None) this Resource was loaded with
microversion = None
@ -979,7 +979,7 @@ class Resource(dict):
:return: A dictionary of key/value pairs where keys are named
as they exist as attributes of this class.
"""
mapping: ty.Union[utils.Munch, dict]
mapping: utils.Munch | dict
if _to_munch:
mapping = utils.Munch()
else:
@ -1073,7 +1073,7 @@ class Resource(dict):
*,
resource_request_key=None,
):
body: ty.Union[dict[str, ty.Any], list[ty.Any]]
body: dict[str, ty.Any] | list[ty.Any]
if patch:
if not self._store_unknown_attrs_as_properties:
# Default case
@ -1250,7 +1250,7 @@ class Resource(dict):
)
@classmethod
def _get_microversion(cls, session: adapter.Adapter) -> ty.Optional[str]:
def _get_microversion(cls, session: adapter.Adapter) -> str | None:
"""Get microversion to use for the given action.
The base version uses the following logic:
@ -1278,10 +1278,10 @@ class Resource(dict):
def _assert_microversion_for(
cls,
session: adapter.Adapter,
expected: ty.Optional[str],
expected: str | None,
*,
error_message: ty.Optional[str] = None,
maximum: ty.Optional[str] = None,
error_message: str | None = None,
maximum: str | None = None,
) -> str:
"""Enforce that the microversion for action satisfies the requirement.
@ -1336,11 +1336,11 @@ class Resource(dict):
self,
session: adapter.Adapter,
prepend_key: bool = True,
base_path: ty.Optional[str] = None,
base_path: str | None = None,
*,
resource_request_key: ty.Optional[str] = None,
resource_response_key: ty.Optional[str] = None,
microversion: ty.Optional[str] = None,
resource_request_key: str | None = None,
resource_response_key: str | None = None,
microversion: str | None = None,
**params: ty.Any,
) -> ty_ext.Self:
"""Create a remote resource based on this instance.
@ -1438,9 +1438,9 @@ class Resource(dict):
session: adapter.Adapter,
data: list[dict[str, ty.Any]],
prepend_key: bool = True,
base_path: ty.Optional[str] = None,
base_path: str | None = None,
*,
microversion: ty.Optional[str] = None,
microversion: str | None = None,
**params: ty.Any,
) -> ty.Generator[ty_ext.Self, None, None]:
"""Create multiple remote resources based on this class and data.
@ -1505,7 +1505,7 @@ class Resource(dict):
)
_body.append(request.body)
body: ty.Union[dict[str, ty.Any], list[ty.Any]] = _body
body: dict[str, ty.Any] | list[ty.Any] = _body
if prepend_key:
if not cls.resources_key:
@ -1559,12 +1559,12 @@ class Resource(dict):
self,
session: adapter.Adapter,
requires_id: bool = True,
base_path: ty.Optional[str] = None,
error_message: ty.Optional[str] = None,
base_path: str | None = None,
error_message: str | None = None,
skip_cache: bool = False,
*,
resource_response_key: ty.Optional[str] = None,
microversion: ty.Optional[str] = None,
resource_response_key: str | None = None,
microversion: str | None = None,
**params: ty.Any,
) -> ty_ext.Self:
"""Get a remote resource based on this instance.
@ -1620,9 +1620,9 @@ class Resource(dict):
def head(
self,
session: adapter.Adapter,
base_path: ty.Optional[str] = None,
base_path: str | None = None,
*,
microversion: ty.Optional[str] = None,
microversion: str | None = None,
) -> ty_ext.Self:
"""Get headers from a remote resource based on this instance.
@ -1665,10 +1665,10 @@ class Resource(dict):
session: adapter.Adapter,
prepend_key: bool = True,
has_body: bool = True,
retry_on_conflict: ty.Optional[bool] = None,
base_path: ty.Optional[str] = None,
retry_on_conflict: bool | None = None,
base_path: str | None = None,
*,
microversion: ty.Optional[str] = None,
microversion: str | None = None,
**kwargs: ty.Any,
) -> ty_ext.Self:
"""Commit the state of the instance to the remote resource.
@ -1857,9 +1857,9 @@ class Resource(dict):
def delete(
self,
session: adapter.Adapter,
error_message: ty.Optional[str] = None,
error_message: str | None = None,
*,
microversion: ty.Optional[str] = None,
microversion: str | None = None,
**kwargs: ty.Any,
) -> ty_ext.Self:
"""Delete the remote resource based on this instance.
@ -1909,11 +1909,11 @@ class Resource(dict):
cls,
session: adapter.Adapter,
paginated: bool = True,
base_path: ty.Optional[str] = None,
base_path: str | None = None,
allow_unknown_params: bool = False,
*,
microversion: ty.Optional[str] = None,
headers: ty.Optional[dict[str, str]] = None,
microversion: str | None = None,
headers: dict[str, str] | None = None,
**params: ty.Any,
) -> ty.Generator[ty_ext.Self, None, None]:
"""This method is a generator which yields resource objects.
@ -2186,12 +2186,12 @@ class Resource(dict):
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[True] = True,
list_base_path: ty.Optional[str] = None,
list_base_path: str | None = None,
*,
microversion: ty.Optional[str] = None,
all_projects: ty.Optional[bool] = None,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty.Optional[ty_ext.Self]: ...
) -> ty_ext.Self | None: ...
@ty.overload
@classmethod
@ -2200,10 +2200,10 @@ class Resource(dict):
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[False],
list_base_path: ty.Optional[str] = None,
list_base_path: str | None = None,
*,
microversion: ty.Optional[str] = None,
all_projects: ty.Optional[bool] = None,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self: ...
@ -2216,12 +2216,12 @@ class Resource(dict):
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool,
list_base_path: ty.Optional[str] = None,
list_base_path: str | None = None,
*,
microversion: ty.Optional[str] = None,
all_projects: ty.Optional[bool] = None,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty.Optional[ty_ext.Self]: ...
) -> ty_ext.Self | None: ...
@classmethod
def find(
@ -2229,12 +2229,12 @@ class Resource(dict):
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool = True,
list_base_path: ty.Optional[str] = None,
list_base_path: str | None = None,
*,
microversion: ty.Optional[str] = None,
all_projects: ty.Optional[bool] = None,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty.Optional[ty_ext.Self]:
) -> ty_ext.Self | None:
"""Find a resource by its name or id.
:param session: The session to use for making this request.
@ -2311,7 +2311,7 @@ class Resource(dict):
)
def _normalize_status(status: ty.Optional[str]) -> ty.Optional[str]:
def _normalize_status(status: str | None) -> str | None:
if status is not None:
status = status.lower()
return status
@ -2321,11 +2321,11 @@ def wait_for_status(
session: adapter.Adapter,
resource: ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> ResourceT:
"""Wait for the resource to be in a particular status.
@ -2399,9 +2399,9 @@ def wait_for_status(
def wait_for_delete(
session: adapter.Adapter,
resource: ResourceT,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
callback: ty.Optional[ty.Callable[[int], None]] = None,
interval: int | float | None = 2,
wait: int | None = None,
callback: ty.Callable[[int], None] | None = None,
) -> ResourceT:
"""Wait for a resource to be deleted.

7
openstack/service_description.py
View File

@ -11,7 +11,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import typing as ty
import warnings
import os_service_types
@ -54,10 +53,8 @@ class ServiceDescription:
def __init__(
self,
service_type: str,
supported_versions: ty.Optional[
dict[str, type[proxy_mod.Proxy]]
] = None,
aliases: ty.Optional[list[str]] = None,
supported_versions: dict[str, type[proxy_mod.Proxy]] | None = None,
aliases: list[str] | None = None,
):
"""Class describing how to interact with a REST service.

10
openstack/shared_file_system/v2/_proxy.py
View File

@ -1191,11 +1191,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -1231,7 +1231,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

6
openstack/test/fakes.py
View File

@ -59,7 +59,7 @@ def generate_fake_resource(
base_attrs: dict[str, ty.Any] = {}
for name, value in inspect.getmembers(
resource_type,
predicate=lambda x: isinstance(x, (fields.Body, fields.URI)),
predicate=lambda x: isinstance(x, fields.Body | fields.URI),
):
if isinstance(value, fields.Body):
target_type = value.type
@ -131,7 +131,7 @@ def generate_fake_resource(
def generate_fake_resources(
resource_type: type[resource.ResourceT],
count: int = 1,
attrs: ty.Optional[dict[str, ty.Any]] = None,
attrs: dict[str, ty.Any] | None = None,
) -> Generator[resource.ResourceT, None, None]:
"""Generate a given number of fake resource entities
@ -161,7 +161,7 @@ def generate_fake_resources(
# (better) type annotations
def generate_fake_proxy(
service: type[service_description.ServiceDescription],
api_version: ty.Optional[str] = None,
api_version: str | None = None,
) -> proxy.Proxy:
"""Generate a fake proxy for the given service type

3
openstack/tests/functional/baremetal/base.py
View File

@ -10,13 +10,12 @@
# License for the specific language governing permissions and limitations
# under the License.
import typing as ty
from openstack.tests.functional import base
class BaseBaremetalTest(base.BaseFunctionalTest):
min_microversion: ty.Optional[str] = None
min_microversion: str | None = None
node_id: str
def setUp(self):

3
openstack/tests/functional/shared_file_system/base.py
View File

@ -10,14 +10,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import typing as ty
from openstack import resource
from openstack.tests.functional import base
class BaseSharedFileSystemTest(base.BaseFunctionalTest):
min_microversion: ty.Optional[str] = None
min_microversion: str | None = None
def setUp(self):
super().setUp()

24
openstack/utils.py
View File

@ -27,7 +27,7 @@ from openstack import _log
from openstack import exceptions
def urljoin(*args: ty.Optional[str]) -> str:
def urljoin(*args: str | None) -> str:
"""A custom version of urljoin that simply joins strings into a path.
The real urljoin takes into account web semantics like when joining a url
@ -38,9 +38,9 @@ def urljoin(*args: ty.Optional[str]) -> str:
def iterate_timeout(
timeout: ty.Optional[int],
timeout: int | None,
message: str,
wait: ty.Union[int, float, None] = 2,
wait: int | float | None = 2,
) -> ty.Generator[int, None, None]:
"""Iterate and raise an exception on timeout.
@ -163,9 +163,7 @@ def supports_version(
def supports_microversion(
adapter: ks_adapter.Adapter,
microversion: ty.Union[
str, int, float, ty.Iterable[ty.Union[str, int, float]]
],
microversion: str | int | float | ty.Iterable[str | int | float],
raise_exception: bool = False,
) -> bool:
"""Determine if the given adapter supports the given microversion.
@ -238,7 +236,7 @@ def require_microversion(adapter: ks_adapter.Adapter, required: str) -> None:
def pick_microversion(
session: ks_adapter.Adapter, required: str
) -> ty.Optional[str]:
) -> str | None:
"""Get a new microversion if it is higher than session's default.
:param session: The session to use for making this request.
@ -281,8 +279,8 @@ def pick_microversion(
def maximum_supported_microversion(
adapter: ks_adapter.Adapter,
client_maximum: ty.Optional[str],
) -> ty.Optional[str]:
client_maximum: str | None,
) -> str | None:
"""Determine the maximum microversion supported by both client and server.
:param adapter: :class:`~keystoneauth1.adapter.Adapter` instance.
@ -331,8 +329,8 @@ def maximum_supported_microversion(
def _hashes_up_to_date(
md5: ty.Optional[str],
sha256: ty.Optional[str],
md5: str | None,
sha256: str | None,
md5_key: str,
sha256_key: str,
) -> bool:
@ -354,7 +352,7 @@ def _hashes_up_to_date(
def _calculate_data_hashes(
data: ty.Union[io.BufferedReader, bytes],
data: io.BufferedReader | bytes,
) -> tuple[str, str]:
_md5 = hashlib.md5(usedforsecurity=False)
_sha256 = hashlib.sha256()
@ -409,7 +407,7 @@ class TinyDAG:
def add_edge(self, u: str, v: str) -> None:
self._graph[u].add(v)
def walk(self, timeout: ty.Optional[int] = None) -> 'TinyDAG':
def walk(self, timeout: int | None = None) -> 'TinyDAG':
"""Start the walking from the beginning."""
if timeout:
self._wait_timeout = timeout

10
openstack/workflow/v2/_proxy.py
View File

@ -301,11 +301,11 @@ class Proxy(proxy.Proxy):
self,
res: resource.ResourceT,
status: str,
failures: ty.Optional[list[str]] = None,
interval: ty.Union[int, float, None] = 2,
wait: ty.Optional[int] = None,
failures: list[str] | None = None,
interval: int | float | None = 2,
wait: int | None = None,
attribute: str = 'status',
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for the resource to be in a particular status.
@ -341,7 +341,7 @@ class Proxy(proxy.Proxy):
res: resource.ResourceT,
interval: int = 2,
wait: int = 120,
callback: ty.Optional[ty.Callable[[int], None]] = None,
callback: ty.Callable[[int], None] | None = None,
) -> resource.ResourceT:
"""Wait for a resource to be deleted.

3
pyproject.toml
View File

@ -1,5 +1,5 @@
[tool.mypy]
python_version = "3.9"
python_version = "3.10"
show_column_numbers = true
show_error_context = true
ignore_missing_imports = true
@ -52,6 +52,7 @@ ignore_errors = true
[tool.ruff]
line-length = 79
target-version = "py310"
[tool.ruff.format]
quote-style = "preserve"