openstack/openstacksdk
Code Issues Proposed changes

typing: Annotate openstack.service_description

Browse Source 
This is rather complicated. [1] is helpful to understand what we're
doing here.

[1] https://adamj.eu/tech/2021/10/18/python-type-hints-how-to-type-a-descriptor/

Change-Id: I51475a28c98906d84d07ebead48cefc998276d81
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane
2025-11-26 21:50:05 +00:00
parent 8334b199c5
commit a17e53ce7a
53 changed files with 225 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
openstack/_services_mixin.py
View File

@@ -1,4 +1,6 @@
# Generated file, to change, run tools/print-services.py
import typing as ty
from openstack import service_description
from openstack.accelerator import accelerator_service
from openstack.baremetal import baremetal_service
@@ -24,6 +26,12 @@ from openstack.placement import placement_service
from openstack.shared_file_system import shared_file_system_service
from openstack.workflow import workflow_service
if ty.TYPE_CHECKING:
# the noqa is necessary as 'proxy' is only referenced in string subscripts
# and ruff doesn't scan for name usage since they're not in annotation
# positions
from openstack import proxy # noqa: F401
class ServicesMixin:
identity = identity_service.IdentityService(service_type='identity')
@@ -46,7 +54,7 @@ class ServicesMixin:
resource_cluster = clustering
cluster = clustering
data_processing = service_description.ServiceDescription(
data_processing = service_description.ServiceDescription['proxy.Proxy'](
service_type='data-processing'
)
@@ -63,17 +71,17 @@ class ServicesMixin:
service_type='key-manager'
)
resource_optimization = service_description.ServiceDescription(
service_type='resource-optimization'
)
resource_optimization = service_description.ServiceDescription[
'proxy.Proxy'
](service_type='resource-optimization')
infra_optim = resource_optimization
message = message_service.MessageService(service_type='message')
messaging = message
application_catalog = service_description.ServiceDescription(
service_type='application-catalog'
)
application_catalog = service_description.ServiceDescription[
'proxy.Proxy'
](service_type='application-catalog')
container_infrastructure_management = container_infrastructure_management_service.ContainerInfrastructureManagementService(
service_type='container-infrastructure-management'
@@ -81,15 +89,19 @@ class ServicesMixin:
container_infra = container_infrastructure_management
container_infrastructure = container_infrastructure_management
search = service_description.ServiceDescription(service_type='search')
search = service_description.ServiceDescription['proxy.Proxy'](
service_type='search'
)
dns = dns_service.DnsService(service_type='dns')
workflow = workflow_service.WorkflowService(service_type='workflow')
rating = service_description.ServiceDescription(service_type='rating')
rating = service_description.ServiceDescription['proxy.Proxy'](
service_type='rating'
)
operator_policy = service_description.ServiceDescription(
operator_policy = service_description.ServiceDescription['proxy.Proxy'](
service_type='operator-policy'
)
policy = operator_policy
@@ -99,9 +111,9 @@ class ServicesMixin:
)
share = shared_file_system
data_protection_orchestration = service_description.ServiceDescription(
service_type='data-protection-orchestration'
)
data_protection_orchestration = service_description.ServiceDescription[
'proxy.Proxy'
](service_type='data-protection-orchestration')
orchestration = orchestration_service.OrchestrationService(
service_type='orchestration'
@@ -113,56 +125,64 @@ class ServicesMixin:
block_store = block_storage
volume = block_storage
alarm = service_description.ServiceDescription(service_type='alarm')
alarm = service_description.ServiceDescription['proxy.Proxy'](
service_type='alarm'
)
alarming = alarm
meter = service_description.ServiceDescription(service_type='meter')
meter = service_description.ServiceDescription['proxy.Proxy'](
service_type='meter'
)
metering = meter
telemetry = meter
event = service_description.ServiceDescription(service_type='event')
event = service_description.ServiceDescription['proxy.Proxy'](
service_type='event'
)
events = event
application_deployment = service_description.ServiceDescription(
service_type='application-deployment'
)
application_deployment = service_description.ServiceDescription[
'proxy.Proxy'
](service_type='application-deployment')
application_deployment = application_deployment
multi_region_network_automation = service_description.ServiceDescription(
service_type='multi-region-network-automation'
)
multi_region_network_automation = service_description.ServiceDescription[
'proxy.Proxy'
](service_type='multi-region-network-automation')
tricircle = multi_region_network_automation
database = database_service.DatabaseService(service_type='database')
application_container = service_description.ServiceDescription(
service_type='application-container'
)
application_container = service_description.ServiceDescription[
'proxy.Proxy'
](service_type='application-container')
container = application_container
root_cause_analysis = service_description.ServiceDescription(
service_type='root-cause-analysis'
)
root_cause_analysis = service_description.ServiceDescription[
'proxy.Proxy'
](service_type='root-cause-analysis')
rca = root_cause_analysis
nfv_orchestration = service_description.ServiceDescription(
nfv_orchestration = service_description.ServiceDescription['proxy.Proxy'](
service_type='nfv-orchestration'
)
network = network_service.NetworkService(service_type='network')
backup = service_description.ServiceDescription(service_type='backup')
backup = service_description.ServiceDescription['proxy.Proxy'](
service_type='backup'
)
monitoring_logging = service_description.ServiceDescription(
monitoring_logging = service_description.ServiceDescription['proxy.Proxy'](
service_type='monitoring-logging'
)
monitoring_log_api = monitoring_logging
monitoring = service_description.ServiceDescription(
monitoring = service_description.ServiceDescription['proxy.Proxy'](
service_type='monitoring'
)
monitoring_events = service_description.ServiceDescription(
monitoring_events = service_description.ServiceDescription['proxy.Proxy'](
service_type='monitoring-events'
)
@@ -173,11 +193,11 @@ class ServicesMixin:
)
ha = instance_ha
reservation = service_description.ServiceDescription(
reservation = service_description.ServiceDescription['proxy.Proxy'](
service_type='reservation'
)
function_engine = service_description.ServiceDescription(
function_engine = service_description.ServiceDescription['proxy.Proxy'](
service_type='function-engine'
)
@@ -185,7 +205,7 @@ class ServicesMixin:
service_type='accelerator'
)
admin_logic = service_description.ServiceDescription(
admin_logic = service_description.ServiceDescription['proxy.Proxy'](
service_type='admin-logic'
)
registration = admin_logic

8
openstack/accelerator/accelerator_service.py
View File

@@ -10,13 +10,15 @@
# License for the specific language governing permissions and limitations
# under the License.
from openstack.accelerator.v2 import _proxy as _proxy_v2
from openstack.accelerator.v2 import _proxy
from openstack import service_description
class AcceleratorService(service_description.ServiceDescription):
class AcceleratorService(
service_description.ServiceDescription[_proxy.Proxy],
):
"""The accelerator service."""
supported_versions = {
'2': _proxy_v2.Proxy,
'2': _proxy.Proxy,
}

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

@@ -22,7 +22,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
# ========== Deployables ==========

2
openstack/baremetal/baremetal_service.py
View File

@@ -14,7 +14,7 @@ from openstack.baremetal.v1 import _proxy
from openstack import service_description
class BaremetalService(service_description.ServiceDescription):
class BaremetalService(service_description.ServiceDescription[_proxy.Proxy]):
"""The bare metal service."""
supported_versions = {

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

@@ -34,7 +34,7 @@ from openstack import utils
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
retriable_status_codes = _common.RETRIABLE_STATUS_CODES

4
openstack/baremetal_introspection/baremetal_introspection_service.py
View File

@@ -14,7 +14,9 @@ from openstack.baremetal_introspection.v1 import _proxy
from openstack import service_description
class BaremetalIntrospectionService(service_description.ServiceDescription):
class BaremetalIntrospectionService(
service_description.ServiceDescription[_proxy.Proxy]
):
"""The bare metal introspection service."""
supported_versions = {

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

@@ -27,7 +27,7 @@ _logger = _log.setup_logging('openstack')
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
_resource_registry = {
"introspection": _introspect.Introspection,

4
openstack/block_storage/block_storage_service.py
View File

@@ -15,7 +15,9 @@ from openstack.block_storage.v3 import _proxy as _v3_proxy
from openstack import service_description
class BlockStorageService(service_description.ServiceDescription):
class BlockStorageService(
service_description.ServiceDescription[_v2_proxy.Proxy | _v3_proxy.Proxy]
):
"""The block storage service."""
supported_versions = {

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

@@ -34,7 +34,7 @@ from openstack import warnings as os_warnings
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
# ========== Extensions ==========

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

@@ -42,7 +42,7 @@ from openstack import warnings as os_warnings
class Proxy(proxy.Proxy):
api_version = '3'
api_version: ty.ClassVar[ty.Literal['3']] = '3'
_resource_registry = {
"availability_zone": availability_zone.AvailabilityZone,

3
openstack/cloud/openstackcloud.py
View File

@@ -670,11 +670,12 @@ class _OpenStackCloudMixin(_services_mixin.ServicesMixin):
# User used string notation. Try to find proper
# resource
(service_name, resource_name) = resource_type.split('.')
service_name, resource_name = resource_type.split('.')
if not hasattr(self, service_name):
raise exceptions.SDKException(
f"service {service_name} is not existing/enabled"
)
service_proxy = getattr(self, service_name)
try:
resource_type = service_proxy._resource_registry[resource_name]

2
openstack/clustering/clustering_service.py
View File

@@ -14,7 +14,7 @@ from openstack.clustering.v1 import _proxy
from openstack import service_description
class ClusteringService(service_description.ServiceDescription):
class ClusteringService(service_description.ServiceDescription[_proxy.Proxy]):
"""The clustering service."""
supported_versions = {

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

@@ -30,7 +30,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
_resource_registry = {
"action": _action.Action,

2
openstack/compute/compute_service.py
View File

@@ -14,7 +14,7 @@ from openstack.compute.v2 import _proxy
from openstack import service_description
class ComputeService(service_description.ServiceDescription):
class ComputeService(service_description.ServiceDescription[_proxy.Proxy]):
"""The compute service."""
supported_versions = {

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

@@ -49,7 +49,7 @@ from openstack import warnings as os_warnings
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
_resource_registry = {
"aggregate": _aggregate.Aggregate,

23
openstack/config/cloud_region.py
View File

@@ -1010,11 +1010,29 @@ class CloudRegion:
endpoint = parse.urljoin(endpoint, 'v2.0')
return endpoint
@ty.overload
def get_session_client(
self,
service_type: str,
version: str | None = None,
constructor: type[proxy.Proxy] = proxy.Proxy,
constructor: None = None,
**kwargs: ty.Any,
) -> proxy.Proxy: ...
@ty.overload
def get_session_client(
self,
service_type: str,
version: str | None = None,
constructor: type[proxy.ProxyT] = ...,
**kwargs: ty.Any,
) -> proxy.ProxyT: ...
def get_session_client(
self,
service_type: str,
version: str | None = None,
constructor: type[proxy.Proxy] | None = None,
**kwargs: ty.Any,
) -> proxy.Proxy:
"""Return a prepped keystoneauth Adapter for a given service.
@@ -1030,6 +1048,9 @@ class CloudRegion:
and it will work like you think.
"""
if constructor is None:
constructor = proxy.Proxy
version_request = self._get_version_request(service_type, version)
kwargs.setdefault('region_name', self.get_region_name(service_type))

15
openstack/connection.py
View File

@@ -380,8 +380,9 @@ class Connection(
session: ks_session.Session | None = None,
app_name: str | None = None,
app_version: str | None = None,
extra_services: list[service_description.ServiceDescription]
| None = None,
extra_services: (
list[service_description.ServiceDescription[ty.Any]] | None
) = None,
strict: bool = False,
use_direct_get: bool | None = None,
task_manager: ty.Any = None,
@@ -529,7 +530,7 @@ class Connection(
)
def add_service(
self, service: service_description.ServiceDescription
self, service: service_description.ServiceDescription['proxy.Proxy']
) -> None:
"""Add a service to the Connection.
@@ -550,13 +551,13 @@ class Connection(
# If we don't have a proxy, just instantiate Proxy so that
# we get an adapter.
if isinstance(service, str):
service = service_description.ServiceDescription(service)
service = service_description.ServiceDescription['proxy.Proxy'](
service
)
# Directly invoke descriptor of the ServiceDescription
def getter(self: 'Connection') -> 'proxy.Proxy':
# TODO(stephenfin): Remove ignore once we have typed
# ServiceDescription
return service.__get__(self, service) # type: ignore
return service.__get__(self, type(self))
# Register the ServiceDescription class (as property)
# with every known alias for a "runtime descriptor"

2
openstack/container_infrastructure_management/container_infrastructure_management_service.py
View File

@@ -15,7 +15,7 @@ from openstack import service_description
class ContainerInfrastructureManagementService(
service_description.ServiceDescription,
service_description.ServiceDescription[_proxy.Proxy],
):
"""The container infrastructure management service."""

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

@@ -29,7 +29,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
_resource_registry = {
"cluster": _cluster.Cluster,

2
openstack/database/database_service.py
View File

@@ -14,7 +14,7 @@ from openstack.database.v1 import _proxy
from openstack import service_description
class DatabaseService(service_description.ServiceDescription):
class DatabaseService(service_description.ServiceDescription[_proxy.Proxy]):
"""The database service."""
supported_versions = {

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

@@ -21,7 +21,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
_resource_registry = {
"database": _database.Database,

2
openstack/dns/dns_service.py
View File

@@ -14,7 +14,7 @@ from openstack.dns.v2 import _proxy
from openstack import service_description
class DnsService(service_description.ServiceDescription):
class DnsService(service_description.ServiceDescription[_proxy.Proxy]):
"""The DNS service."""
supported_versions = {

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

@@ -31,7 +31,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
_resource_registry = {
"blacklist": _blacklist.Blacklist,

4
openstack/identity/identity_service.py
View File

@@ -15,7 +15,9 @@ from openstack.identity.v3 import _proxy as _proxy_v3
from openstack import service_description
class IdentityService(service_description.ServiceDescription):
class IdentityService(
service_description.ServiceDescription[_proxy_v2.Proxy | _proxy_v3.Proxy]
):
"""The identity service."""
supported_versions = {

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

@@ -21,7 +21,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
def extensions(self):
"""Retrieve a generator of extensions

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

@@ -64,7 +64,7 @@ from openstack import warnings as os_warnings
class Proxy(proxy.Proxy):
api_version = '3'
api_version: ty.ClassVar[ty.Literal['3']] = '3'
_resource_registry = {
"application_credential": _application_credential.ApplicationCredential, # noqa: E501

4
openstack/image/image_service.py
View File

@@ -15,7 +15,9 @@ from openstack.image.v2 import _proxy as _proxy_v2
from openstack import service_description
class ImageService(service_description.ServiceDescription):
class ImageService(
service_description.ServiceDescription[_proxy_v1.Proxy | _proxy_v2.Proxy]
):
"""The image service."""
supported_versions = {

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

@@ -37,7 +37,7 @@ def _get_name_and_filename(name, image_format):
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
retriable_status_codes = [503]

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

@@ -54,7 +54,7 @@ def _get_name_and_filename(name, image_format):
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
_resource_registry = {
"cache": _cache.Cache,

2
openstack/instance_ha/instance_ha_service.py
View File

@@ -16,7 +16,7 @@ from openstack.instance_ha.v1 import _proxy
from openstack import service_description
class InstanceHaService(service_description.ServiceDescription):
class InstanceHaService(service_description.ServiceDescription[_proxy.Proxy]):
"""The HA service."""
supported_versions = {

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

@@ -24,7 +24,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
_resource_registry = {
"host": _host.Host,

2
openstack/key_manager/key_manager_service.py
View File

@@ -14,7 +14,7 @@ from openstack.key_manager.v1 import _proxy
from openstack import service_description
class KeyManagerService(service_description.ServiceDescription):
class KeyManagerService(service_description.ServiceDescription[_proxy.Proxy]):
"""The key manager service."""
supported_versions = {

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

@@ -22,7 +22,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
_resource_registry = {
"container": _container.Container,

4
openstack/load_balancer/load_balancer_service.py
View File

@@ -14,7 +14,9 @@ from openstack.load_balancer.v2 import _proxy
from openstack import service_description
class LoadBalancerService(service_description.ServiceDescription):
class LoadBalancerService(
service_description.ServiceDescription[_proxy.Proxy]
):
"""The load balancer service."""
supported_versions = {

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

@@ -33,7 +33,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
_resource_registry = {
"amphora": _amphora.Amphora,

2
openstack/message/message_service.py
View File

@@ -14,7 +14,7 @@ from openstack.message.v2 import _proxy
from openstack import service_description
class MessageService(service_description.ServiceDescription):
class MessageService(service_description.ServiceDescription[_proxy.Proxy]):
"""The message service."""
supported_versions = {

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

@@ -21,7 +21,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
_resource_registry = {
"claim": _claim.Claim,

2
openstack/network/network_service.py
View File

@@ -14,7 +14,7 @@ from openstack.network.v2 import _proxy
from openstack import service_description
class NetworkService(service_description.ServiceDescription):
class NetworkService(service_description.ServiceDescription[_proxy.Proxy]):
"""The network service."""
supported_versions = {

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

@@ -109,7 +109,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
_resource_registry = {
"address_group": _address_group.AddressGroup,

2
openstack/object_store/object_store_service.py
View File

@@ -14,7 +14,7 @@ from openstack.object_store.v1 import _proxy
from openstack import service_description
class ObjectStoreService(service_description.ServiceDescription):
class ObjectStoreService(service_description.ServiceDescription[_proxy.Proxy]):
"""The object store service."""
supported_versions = {

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

@@ -43,7 +43,7 @@ def _get_expiration(expiration):
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
_resource_registry = {
"account": _account.Account,

4
openstack/orchestration/orchestration_service.py
View File

@@ -14,7 +14,9 @@ from openstack.orchestration.v1 import _proxy
from openstack import service_description
class OrchestrationService(service_description.ServiceDescription):
class OrchestrationService(
service_description.ServiceDescription[_proxy.Proxy]
):
"""The orchestration service."""
supported_versions = {

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

@@ -30,7 +30,7 @@ from openstack import resource
# TODO(rladntjr4): Some of these methods support lookup by ID, while others
# support lookup by ID or name. We should choose one and use it consistently.
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
_resource_registry = {
"resource": _resource.Resource,
@@ -109,7 +109,9 @@ class Proxy(proxy.Proxy):
)
return stack_attrs
def create_stack(self, preview=False, **attrs):
def create_stack(
self, preview: bool = False, **attrs: ty.Any
) -> _stack.Stack:
"""Create a new stack from attributes
:param bool preview: When ``True``, a preview endpoint will be used to

2
openstack/placement/placement_service.py
View File

@@ -14,7 +14,7 @@ from openstack.placement.v1 import _proxy
from openstack import service_description
class PlacementService(service_description.ServiceDescription):
class PlacementService(service_description.ServiceDescription[_proxy.Proxy]):
"""The placement service."""
supported_versions = {

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

@@ -23,7 +23,7 @@ from openstack import resource
class Proxy(proxy.Proxy):
api_version = '1'
api_version: ty.ClassVar[ty.Literal['1']] = '1'
_resource_registry = {
"resource_class": _resource_class.ResourceClass,

3
openstack/proxy.py
View File

@@ -50,6 +50,9 @@ if ty.TYPE_CHECKING:
from openstack import connection
ProxyT = ty.TypeVar('ProxyT', bound='Proxy')
def normalize_metric_name(name: str) -> str:
name = name.replace('.', '_')
name = name.replace(':', '_')

69
openstack/service_description.py
View File

@@ -11,6 +11,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import typing as ty
import warnings
import os_service_types
@@ -24,16 +25,19 @@ __all__ = [
'ServiceDescription',
]
if ty.TYPE_CHECKING:
from openstack import connection
_logger = _log.setup_logging('openstack')
_service_type_manager = os_service_types.ServiceTypes()
class _ServiceDisabledProxyShim:
def __init__(self, service_type, reason):
def __init__(self, service_type: str, reason: str | None) -> None:
self.service_type = service_type
self.reason = reason
def __getattr__(self, item):
def __getattr__(self, item: ty.Any) -> ty.Any:
raise exceptions.ServiceDisabledException(
"Service '{service_type}' is disabled because its configuration "
"could not be loaded. {reason}".format(
@@ -42,7 +46,7 @@ class _ServiceDisabledProxyShim:
)
class ServiceDescription:
class ServiceDescription(ty.Generic[proxy_mod.ProxyT]):
#: Dictionary of supported versions and proxy classes for that version
supported_versions: dict[str, type[proxy_mod.Proxy]] = {}
#: main service_type to use to find this service in the catalog
@@ -84,17 +88,40 @@ class ServiceDescription:
self.aliases = aliases or self.aliases
self.all_types = [service_type, *self.aliases]
def __get__(self, instance, owner):
@ty.overload
def __get__(self, instance: None, owner: None) -> 'ServiceDescription': ...
# NOTE(stephenfin): We would like to type instance as
# connection.Connection, but due to how we construct that object, we can't
# do so yet.
@ty.overload
def __get__(
self,
instance: ty.Any,
owner: type[object],
) -> proxy_mod.ProxyT: ...
def __get__(
self,
instance: ty.Any,
owner: type[object] | None,
) -> 'ServiceDescription | proxy_mod.ProxyT':
if instance is None:
return self
if self.service_type in instance._proxies:
return instance._proxies[self.service_type]
return ty.cast(
proxy_mod.ProxyT, instance._proxies[self.service_type]
)
proxy = self._make_proxy(instance)
if isinstance(proxy, _ServiceDisabledProxyShim):
instance._proxies[self.service_type] = proxy
return instance._proxies[self.service_type]
return ty.cast(
proxy_mod.ProxyT,
instance._proxies[self.service_type],
)
# The keystone proxy has a method called get_endpoint
# that is about managing keystone endpoints. This is
@@ -152,7 +179,10 @@ class ServiceDescription:
)
)
def _make_proxy(self, instance):
def _make_proxy(
self,
instance: 'connection.Connection',
) -> proxy_mod.ProxyT | proxy_mod.Proxy:
"""Create a Proxy for the service in question.
:param instance: The `openstack.connection.Connection` we're working
@@ -162,9 +192,14 @@ class ServiceDescription:
# This is not a valid service.
if not config.has_service(self.service_type):
return _ServiceDisabledProxyShim(
self.service_type,
config.get_disabled_reason(self.service_type),
# NOTE(stephenfin): Yes, we are lying here. But that's okay: they
# should behave identically in a typing context
return ty.cast(
proxy_mod.ProxyT,
_ServiceDisabledProxyShim(
self.service_type,
config.get_disabled_reason(self.service_type),
),
)
# This is a valid service type, but we don't know anything about it so
@@ -246,11 +281,14 @@ class ServiceDescription:
return proxy_obj
data = proxy_obj.get_endpoint_data()
if not data and instance._strict_proxies:
raise exceptions.ServiceDiscoveryException(
f"Failed to create a working proxy for service "
f"{self.service_type}: No endpoint data found."
)
if not data:
if instance._strict_proxies:
raise exceptions.ServiceDiscoveryException(
f"Failed to create a working proxy for service "
f"{self.service_type}: No endpoint data found."
)
else:
return proxy_obj
# If we've gotten here with a proxy object it means we have
# an endpoint_override in place. If the catalog_url and
@@ -334,6 +372,7 @@ class ServiceDescription:
return config.get_session_client(
self.service_type,
version=version_string,
constructor=proxy_class,
allow_version_hack=True,
max_version=f'{supported_versions[-1]!s}.latest',

4
openstack/shared_file_system/shared_file_system_service.py
View File

@@ -14,7 +14,9 @@ from openstack import service_description
from openstack.shared_file_system.v2 import _proxy
class SharedFilesystemService(service_description.ServiceDescription):
class SharedFilesystemService(
service_description.ServiceDescription[_proxy.Proxy]
):
"""The shared file systems service."""
supported_versions = {

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

@@ -46,7 +46,7 @@ from openstack.shared_file_system.v2 import user_message as _user_message
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
_resource_registry = {
"availability_zone": _availability_zone.AvailabilityZone,

21
openstack/tests/functional/orchestration/v1/test_stack.py
View File

@@ -20,10 +20,7 @@ from openstack.tests.functional.network.v2 import test_network
class TestStack(base.BaseFunctionalTest):
NAME = 'test_stack'
stack = None
network = None
subnet = None
cidr = '10.99.99.0/16'
CIDR = '10.99.99.0/16'
_wait_for_timeout_key = 'OPENSTACKSDK_FUNC_TEST_TIMEOUT_ORCHESTRATION'
@@ -41,7 +38,7 @@ class TestStack(base.BaseFunctionalTest):
# the shade layer.
template['heat_template_version'] = '2013-05-23'
self.network, self.subnet = test_network.create_network(
self.operator_cloud, self.NAME, self.cidr
self.operator_cloud, self.NAME, self.CIDR
)
parameters = {
'image': image.id,
@@ -53,10 +50,10 @@ class TestStack(base.BaseFunctionalTest):
parameters=parameters,
template=template,
)
assert isinstance(sot, stack.Stack)
self.assertEqual(True, (sot.id is not None))
self.stack = sot
self.assertIsInstance(sot, stack.Stack)
self.assertIsNotNone(sot.id)
self.assertEqual(self.NAME, sot.name)
self.stack = sot
self.operator_cloud.orchestration.wait_for_status(
sot,
status='CREATE_COMPLETE',
@@ -92,18 +89,18 @@ class TestStack(base.BaseFunctionalTest):
# when
self.operator_cloud.orchestration.suspend_stack(self.stack)
sot = self.operator_cloud.orchestration.wait_for_status(
self.stack = self.operator_cloud.orchestration.wait_for_status(
self.stack, suspend_status, wait=self._wait_for_timeout
)
# then
self.assertEqual(suspend_status, sot.status)
self.assertEqual(suspend_status, self.stack.status)
# when
self.operator_cloud.orchestration.resume_stack(self.stack)
sot = self.operator_cloud.orchestration.wait_for_status(
self.stack = self.operator_cloud.orchestration.wait_for_status(
self.stack, resume_status, wait=self._wait_for_timeout
)
# then
self.assertEqual(resume_status, sot.status)
self.assertEqual(resume_status, self.stack.status)

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

@@ -20,7 +20,7 @@ from openstack.workflow.v2 import workflow as _workflow
class Proxy(proxy.Proxy):
api_version = '2'
api_version: ty.ClassVar[ty.Literal['2']] = '2'
_resource_registry = {
"execution": _execution.Execution,

4
openstack/workflow/workflow_service.py
View File

@@ -14,7 +14,9 @@ from openstack import service_description
from openstack.workflow.v2 import _proxy
class WorkflowService(service_description.ServiceDescription):
class WorkflowService(
service_description.ServiceDescription[_proxy.Proxy],
):
"""The workflow service."""
supported_versions = {

3
tools/print-services.py
View File

@@ -46,6 +46,9 @@ def make_names():
dm = 'service_description'
dc = desc_class.__name__
if dc == 'ServiceDescription':
dc = '{dc}[proxy.Proxy]'
services.append(
f"{st} = {dm}.{dc}(service_type='{service_type}')",
)