pre-commit: Bump versions

This brings in a new version of mypy which requires a number of changes.

Change-Id: I1bdbde55107ffebe287a7344086c34ebdac734f1
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane
2025-06-09 12:14:24 +01:00
parent a42b3c51f3
commit 66e5e3b59c
18 changed files with 587 additions and 55 deletions

View File

@@ -1,13 +1,13 @@
--- ---
repos: repos:
- repo: https://github.com/pre-commit/pre-commit-hooks - repo: https://github.com/pre-commit/pre-commit-hooks
rev: v5.0.0 rev: v6.0.0
hooks: hooks:
- id: trailing-whitespace - id: trailing-whitespace
- id: mixed-line-ending - id: mixed-line-ending
args: ['--fix', 'lf'] args: ['--fix', 'lf']
exclude: '.*\.(svg)$' exclude: '.*\.(svg)$'
- id: check-byte-order-marker - id: fix-byte-order-marker
- id: check-executables-have-shebangs - id: check-executables-have-shebangs
- id: check-merge-conflict - id: check-merge-conflict
- id: debug-statements - id: debug-statements
@@ -15,11 +15,11 @@ repos:
files: .*\.(yaml|yml)$ files: .*\.(yaml|yml)$
exclude: '^zuul.d/.*$' exclude: '^zuul.d/.*$'
- repo: https://github.com/PyCQA/doc8 - repo: https://github.com/PyCQA/doc8
rev: v1.1.2 rev: v2.0.0
hooks: hooks:
- id: doc8 - id: doc8
- repo: https://github.com/astral-sh/ruff-pre-commit - repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.11.11 rev: v0.12.12
hooks: hooks:
- id: ruff-check - id: ruff-check
args: ['--fix', '--unsafe-fixes'] args: ['--fix', '--unsafe-fixes']
@@ -32,7 +32,7 @@ repos:
- flake8-import-order~=0.18.2 - flake8-import-order~=0.18.2
exclude: '^(doc|releasenotes|tools)/.*$' exclude: '^(doc|releasenotes|tools)/.*$'
- repo: https://github.com/pre-commit/mirrors-mypy - repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.15.0 rev: v1.17.1
hooks: hooks:
- id: mypy - id: mypy
additional_dependencies: additional_dependencies:

View File

@@ -10,6 +10,10 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import typing as ty
from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack import exceptions from openstack import exceptions
from openstack import resource from openstack import resource
@@ -52,9 +56,72 @@ class Service(resource.Resource):
#: The date and time when the resource was updated #: The date and time when the resource was updated
updated_at = resource.Body('updated_at') updated_at = resource.Body('updated_at')
@ty.overload
@classmethod @classmethod
def find(cls, session, name_or_id, ignore_missing=True, **params): def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[True] = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[False],
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self: ...
# excuse the duplication here: it's mypy's fault
# https://github.com/python/mypy/issues/14764
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None:
# No direct request possible, thus go directly to list # No direct request possible, thus go directly to list
if list_base_path:
params['base_path'] = list_base_path
# all_projects is a special case that is used by multiple services. We
# handle it here since it doesn't make sense to pass it to the .fetch
# call above
if all_projects is not None:
params['all_projects'] = all_projects
data = cls.list(session, **params) data = cls.list(session, **params)
result = None result = None

View File

@@ -13,7 +13,8 @@
import enum import enum
import typing as ty import typing as ty
from keystoneauth1 import adapter as ksa_adapter from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack import exceptions from openstack import exceptions
from openstack import resource from openstack import resource
@@ -95,9 +96,72 @@ class Service(resource.Resource):
# 3.32 introduced the 'set-log' action # 3.32 introduced the 'set-log' action
_max_microversion = '3.32' _max_microversion = '3.32'
@ty.overload
@classmethod @classmethod
def find(cls, session, name_or_id, ignore_missing=True, **params): def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[True] = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[False],
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self: ...
# excuse the duplication here: it's mypy's fault
# https://github.com/python/mypy/issues/14764
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None:
# No direct request possible, thus go directly to list # No direct request possible, thus go directly to list
if list_base_path:
params['base_path'] = list_base_path
# all_projects is a special case that is used by multiple services. We
# handle it here since it doesn't make sense to pass it to the .fetch
# call above
if all_projects is not None:
params['all_projects'] = all_projects
data = cls.list(session, **params) data = cls.list(session, **params)
result = None result = None
@@ -172,7 +236,7 @@ class Service(resource.Resource):
@classmethod @classmethod
def set_log_levels( def set_log_levels(
cls, cls,
session: ksa_adapter.Adapter, session: adapter.Adapter,
*, *,
level: Level, level: Level,
binary: Binary | None = None, binary: Binary | None = None,
@@ -207,7 +271,7 @@ class Service(resource.Resource):
@classmethod @classmethod
def get_log_levels( def get_log_levels(
cls, cls,
session: ksa_adapter.Adapter, session: adapter.Adapter,
*, *,
binary: Binary | None = None, binary: Binary | None = None,
server: str | None = None, server: str | None = None,

View File

@@ -73,6 +73,13 @@ class _OpenStackCloudMixin(_services_mixin.ServicesMixin):
config: cloud_region.CloudRegion config: cloud_region.CloudRegion
cache_enabled: bool
_cache_expirations: dict[str, int]
_cache: 'cache_region.CacheRegion'
verify: bool | str | None
cert: str | tuple[str, str] | None
def __init__( def __init__(
self, self,
cloud: str | None = None, cloud: str | None = None,
@@ -228,7 +235,7 @@ class _OpenStackCloudMixin(_services_mixin.ServicesMixin):
self.default_interface = self.config.get_interface() self.default_interface = self.config.get_interface()
self.force_ipv4 = self.config.force_ipv4 self.force_ipv4 = self.config.force_ipv4
(self.verify, self.cert) = self.config.get_requests_verify_args() self.verify, self.cert = self.config.get_requests_verify_args()
# Turn off urllib3 warnings about insecure certs if we have # Turn off urllib3 warnings about insecure certs if we have
# explicitly configured requests to tell it we do not want # explicitly configured requests to tell it we do not want

View File

@@ -142,7 +142,7 @@ class QuotaSet(resource.Resource):
self._header.attributes.update(headers) self._header.attributes.update(headers)
self._header.clean() self._header.clean()
self._update_location() self._update_location()
dict.update(self, self.to_dict()) # type: ignore dict.update(self, self.to_dict())
def _prepare_request_body( def _prepare_request_body(
self, self,

View File

@@ -10,6 +10,11 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import typing as ty
from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack import exceptions from openstack import exceptions
from openstack import resource from openstack import resource
from openstack import utils from openstack import utils
@@ -91,11 +96,16 @@ class Flavor(resource.Resource):
@classmethod @classmethod
def list( def list(
cls, cls,
session, session: adapter.Adapter,
paginated=True, paginated: bool = True,
base_path='/flavors/detail', base_path: str | None = '/flavors/detail',
**params, allow_unknown_params: bool = False,
): *,
microversion: str | None = None,
headers: dict[str, str] | None = None,
max_items: int | None = None,
**params: ty.Any,
) -> ty.Generator[ty_ext.Self, None, None]:
# Find will invoke list when name was passed. Since we want to return # Find will invoke list when name was passed. Since we want to return
# flavor with details (same as direct get) we need to swap default here # flavor with details (same as direct get) we need to swap default here
# and list with "/flavors" if no details explicitely requested # and list with "/flavors" if no details explicitely requested
@@ -104,7 +114,14 @@ class Flavor(resource.Resource):
# Force it to string to avoid requests skipping it. # Force it to string to avoid requests skipping it.
params['is_public'] = 'None' params['is_public'] = 'None'
return super().list( return super().list(
session, paginated=paginated, base_path=base_path, **params session,
paginated=paginated,
base_path=base_path,
allow_unknown_params=allow_unknown_params,
microversion=microversion,
headers=headers,
max_items=max_items,
**params,
) )
def _action(self, session, body, microversion=None): def _action(self, session, body, microversion=None):

View File

@@ -10,6 +10,11 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import typing as ty
from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack import resource from openstack import resource
from openstack import utils from openstack import utils
@@ -34,13 +39,18 @@ class ServerIP(resource.Resource):
@classmethod @classmethod
def list( def list(
cls, cls,
session, session: adapter.Adapter,
paginated=False, paginated: bool = False,
server_id=None, base_path: str | None = None,
network_label=None, allow_unknown_params: bool = False,
base_path=None, *,
**params, microversion: str | None = None,
): headers: dict[str, str] | None = None,
max_items: int | None = None,
server_id: str | None = None,
network_label: str | None = None,
**params: ty.Any,
) -> ty.Generator[ty_ext.Self, None, None]:
if base_path is None: if base_path is None:
base_path = cls.base_path base_path = cls.base_path

View File

@@ -10,6 +10,11 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import typing as ty
from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack import exceptions from openstack import exceptions
from openstack import resource from openstack import resource
from openstack import utils from openstack import utils
@@ -55,10 +60,76 @@ class Service(resource.Resource):
_max_microversion = '2.69' _max_microversion = '2.69'
@ty.overload
@classmethod @classmethod
def find(cls, session, name_or_id, ignore_missing=True, **params): def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[True] = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[False],
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self: ...
# excuse the duplication here: it's mypy's fault
# https://github.com/python/mypy/issues/14764
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None:
# No direct request possible, thus go directly to list # No direct request possible, thus go directly to list
data = cls.list(session, **params) if list_base_path:
params['base_path'] = list_base_path
# all_projects is a special case that is used by multiple services. We
# handle it here since it doesn't make sense to pass it to the .fetch
# call above
if all_projects is not None:
params['all_projects'] = all_projects
data = cls.list(
session,
base_path=list_base_path,
)
result = None result = None
for maybe_result in data: for maybe_result in data:

View File

@@ -434,7 +434,9 @@ class CloudRegion:
"""Sets the Session constructor.""" """Sets the Session constructor."""
self._session_constructor = session_constructor self._session_constructor = session_constructor
def get_requests_verify_args(self) -> tuple[bool, str | None]: def get_requests_verify_args(
self,
) -> tuple[bool | str | None, str | tuple[str, str] | None]:
"""Return the verify and cert values for the requests library.""" """Return the verify and cert values for the requests library."""
insecure = self.config.get('insecure', False) insecure = self.config.get('insecure', False)
verify = self.config.get('verify', True) verify = self.config.get('verify', True)

View File

@@ -10,15 +10,73 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import typing as ty
import urllib.parse import urllib.parse
from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack import exceptions from openstack import exceptions
from openstack import resource from openstack import resource
class Resource(resource.Resource): class Resource(resource.Resource):
@ty.overload
@classmethod @classmethod
def find(cls, session, name_or_id, ignore_missing=True, **params): def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[True] = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[False],
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self: ...
# excuse the duplication here: it's mypy's fault
# https://github.com/python/mypy/issues/14764
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None:
"""Find a resource by its name or id. """Find a resource by its name or id.
:param session: The session to use for making this request. :param session: The session to use for making this request.
@@ -46,7 +104,9 @@ class Resource(resource.Resource):
# Try to short-circuit by looking directly for a matching ID. # Try to short-circuit by looking directly for a matching ID.
try: try:
match = cls.existing( match = cls.existing(
id=name_or_id, connection=session._get_connection(), **params id=name_or_id,
connection=session._get_connection(), # type: ignore
**params,
) )
return match.fetch(session) return match.fetch(session)
except exceptions.SDKException: except exceptions.SDKException:
@@ -59,7 +119,13 @@ class Resource(resource.Resource):
): ):
params['name'] = name_or_id params['name'] = name_or_id
data = cls.list(session, **params) data = cls.list(
session,
list_base_path=list_base_path,
microversion=microversion,
all_projects=all_projects,
**params,
)
result = cls._get_one_match(name_or_id, data) result = cls._get_one_match(name_or_id, data)
if result is not None: if result is not None:
@@ -74,16 +140,21 @@ class Resource(resource.Resource):
@classmethod @classmethod
def list( def list(
cls, cls,
session, session: adapter.Adapter,
project_id=None, paginated: bool = True,
all_projects=None, base_path: str | None = None,
**params, allow_unknown_params: bool = False,
): *,
headers: dict[str, str] | None = ( microversion: str | None = None,
{} if project_id or all_projects else None headers: dict[str, str] | None = None,
) max_items: int | None = None,
project_id: str | None = None,
if headers is not None: all_projects: bool | None = None,
**params: ty.Any,
) -> ty.Generator[ty_ext.Self, None, None]:
if project_id or all_projects is not None:
if headers is None:
headers = {}
if project_id: if project_id:
headers["x-auth-sudo-project-id"] = str(project_id) headers["x-auth-sudo-project-id"] = str(project_id)
if all_projects: if all_projects:

View File

@@ -10,6 +10,11 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import typing as ty
from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack import resource from openstack import resource
@@ -43,7 +48,18 @@ class Extension(resource.Resource):
updated_at = resource.Body('updated') updated_at = resource.Body('updated')
@classmethod @classmethod
def list(cls, session, paginated=False, base_path=None, **params): def list(
cls,
session: adapter.Adapter,
paginated: bool = True,
base_path: str | None = None,
allow_unknown_params: bool = False,
*,
microversion: str | None = None,
headers: dict[str, str] | None = None,
max_items: int | None = None,
**params: ty.Any,
) -> ty.Generator[ty_ext.Self, None, None]:
if base_path is None: if base_path is None:
base_path = cls.base_path base_path = cls.base_path

View File

@@ -150,4 +150,4 @@ class Limit(resource.Resource):
self._header.attributes.update(headers) self._header.attributes.update(headers)
self._header.clean() self._header.clean()
self._update_location() self._update_location()
dict.update(self, self.to_dict()) # type: ignore dict.update(self, self.to_dict())

View File

@@ -149,4 +149,4 @@ class RegisteredLimit(resource.Resource):
self._header.attributes.update(headers) self._header.attributes.update(headers)
self._header.clean() self._header.clean()
self._update_location() self._update_location()
dict.update(self, self.to_dict()) # type: ignore dict.update(self, self.to_dict())

View File

@@ -10,6 +10,11 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import typing as ty
from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack import resource from openstack import resource
@@ -27,7 +32,17 @@ class Version(resource.Resource):
updated = resource.Body('updated') updated = resource.Body('updated')
@classmethod @classmethod
def list(cls, session, paginated=False, base_path=None, **params): def list(
cls,
session: adapter.Adapter,
paginated: bool = True,
base_path: str | None = None,
allow_unknown_params: bool = False,
*,
microversion: str | None = None,
headers: dict[str, str] | None = None,
**params: ty.Any,
) -> ty.Generator[ty_ext.Self, None, None]:
if base_path is None: if base_path is None:
base_path = cls.base_path base_path = cls.base_path

View File

@@ -10,6 +10,11 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import typing as ty
from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack import exceptions from openstack import exceptions
from openstack.image import _download from openstack.image import _download
from openstack import resource from openstack import resource
@@ -87,8 +92,62 @@ class Image(resource.Resource, _download.DownloadMixin):
#: The timestamp when this image was last updated. #: The timestamp when this image was last updated.
updated_at = resource.Body('updated_at') updated_at = resource.Body('updated_at')
@ty.overload
@classmethod @classmethod
def find(cls, session, name_or_id, ignore_missing=True, **params): def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[True] = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[False],
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self: ...
# excuse the duplication here: it's mypy's fault
# https://github.com/python/mypy/issues/14764
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None:
"""Find a resource by its name or id. """Find a resource by its name or id.
:param session: The session to use for making this request. :param session: The session to use for making this request.
@@ -117,7 +176,7 @@ class Image(resource.Resource, _download.DownloadMixin):
try: try:
match = cls.existing( match = cls.existing(
id=name_or_id, id=name_or_id,
connection=session._get_connection(), connection=session._get_connection(), # type: ignore
**params, **params,
) )
return match.fetch(session, **params) return match.fetch(session, **params)
@@ -126,7 +185,12 @@ class Image(resource.Resource, _download.DownloadMixin):
params['name'] = name_or_id params['name'] = name_or_id
data = cls.list(session, base_path='/images/detail', **params) data = cls.list(
session,
base_path='/images/detail',
all_projects=all_projects,
**params,
)
result = cls._get_one_match(name_or_id, data) result = cls._get_one_match(name_or_id, data)
if result is not None: if result is not None:

View File

@@ -12,6 +12,9 @@
import typing as ty import typing as ty
from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack.common import tag from openstack.common import tag
from openstack import exceptions from openstack import exceptions
from openstack.image import _download from openstack.image import _download
@@ -403,10 +406,72 @@ class Image(resource.Resource, tag.TagMixin, _download.DownloadMixin):
return request return request
@ty.overload
@classmethod @classmethod
def find(cls, session, name_or_id, ignore_missing=True, **params): def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[True] = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[False],
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self: ...
# excuse the duplication here: it's mypy's fault
# https://github.com/python/mypy/issues/14764
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None:
# Do a regular search first (ignoring missing) # Do a regular search first (ignoring missing)
result = super().find(session, name_or_id, True, **params) result = super().find(
session,
name_or_id,
ignore_missing=True,
list_base_path=list_base_path,
microversion=microversion,
all_projects=all_projects,
**params,
)
if result: if result:
return result return result

View File

@@ -9,6 +9,12 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import typing as ty
from keystoneauth1 import adapter
import typing_extensions as ty_ext
from openstack.common import tag from openstack.common import tag
from openstack import exceptions from openstack import exceptions
from openstack import resource from openstack import resource
@@ -246,8 +252,62 @@ class Stack(resource.Resource):
raise exceptions.NotFoundException(f"No stack found for {self.id}") raise exceptions.NotFoundException(f"No stack found for {self.id}")
return self return self
@ty.overload
@classmethod @classmethod
def find(cls, session, name_or_id, ignore_missing=True, **params): def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[True] = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: ty.Literal[False],
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self: ...
# excuse the duplication here: it's mypy's fault
# https://github.com/python/mypy/issues/14764
@ty.overload
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None: ...
@classmethod
def find(
cls,
session: adapter.Adapter,
name_or_id: str,
ignore_missing: bool = True,
list_base_path: str | None = None,
*,
microversion: str | None = None,
all_projects: bool | None = None,
**params: ty.Any,
) -> ty_ext.Self | None:
"""Find a resource by its name or id. """Find a resource by its name or id.
:param session: The session to use for making this request. :param session: The session to use for making this request.
@@ -275,7 +335,9 @@ class Stack(resource.Resource):
# Try to short-circuit by looking directly for a matching ID. # Try to short-circuit by looking directly for a matching ID.
try: try:
match = cls.existing( match = cls.existing(
id=name_or_id, connection=session._get_connection(), **params id=name_or_id,
connection=session._get_connection(), # type: ignore
**params,
) )
return match.fetch(session, **params) return match.fetch(session, **params)
except exceptions.NotFoundException: except exceptions.NotFoundException:
@@ -286,6 +348,7 @@ class Stack(resource.Resource):
if ignore_missing: if ignore_missing:
return None return None
raise exceptions.NotFoundException( raise exceptions.NotFoundException(
f"No {cls.__name__} found for {name_or_id}" f"No {cls.__name__} found for {name_or_id}"
) )

View File

@@ -512,7 +512,7 @@ class Resource(dict):
# obj.items() ... but I think the if not obj: is short-circuiting down # obj.items() ... but I think the if not obj: is short-circuiting down
# in the C code and thus since we don't store the data in self[] it's # in the C code and thus since we don't store the data in self[] it's
# always False even if we override __len__ or __bool__. # always False even if we override __len__ or __bool__.
dict.update(self, self.to_dict()) # type: ignore dict.update(self, self.to_dict())
@classmethod @classmethod
def _attributes_iterator( def _attributes_iterator(
@@ -703,7 +703,7 @@ class Resource(dict):
# obj.items() ... but I think the if not obj: is short-circuiting down # obj.items() ... but I think the if not obj: is short-circuiting down
# in the C code and thus since we don't store the data in self[] it's # in the C code and thus since we don't store the data in self[] it's
# always False even if we override __len__ or __bool__. # always False even if we override __len__ or __bool__.
dict.update(self, self.to_dict()) # type: ignore dict.update(self, self.to_dict())
def _collect_attrs(self, attrs): def _collect_attrs(self, attrs):
"""Given attributes, return a dict per type of attribute """Given attributes, return a dict per type of attribute
@@ -1268,7 +1268,7 @@ class Resource(dict):
self._header.attributes.update(headers) self._header.attributes.update(headers)
self._header.clean() self._header.clean()
self._update_location() self._update_location()
dict.update(self, self.to_dict()) # type: ignore dict.update(self, self.to_dict())
@classmethod @classmethod
def _get_session(cls, session: AdapterT) -> AdapterT: def _get_session(cls, session: AdapterT) -> AdapterT: