openstack/os-service-types
Code Issues Proposed changes

Add typing, mypy

Browse Source 
Change-Id: I2ad60cfdd58b21a9b43063c7f87e2084a0086cc5
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane
2025-06-27 09:38:49 +01:00
parent 2f112eac23
commit 0cd97ec814
11 changed files with 174 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
.pre-commit-config.yaml
View File

@@ -26,3 +26,15 @@ repos:
- id: hacking
exclude: '^(doc|releasenotes|tools)/.*$'
additional_dependencies: []
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.15.0
hooks:
- id: mypy
# keep this in-sync with '[mypy] exclude' in 'pyproject.toml'
exclude: |
(?x)(
doc/.*
| releasenotes/.*
)
additional_dependencies:
- types-requests

22
os_service_types/__init__.py
View File

@@ -9,17 +9,27 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
__all__ = ['__version__', 'ServiceTypes']
from typing import TYPE_CHECKING, Optional
import pbr.version
from os_service_types.service_types import ServiceTypes # flake8: noqa
from os_service_types.service_types import ServiceTypes
if TYPE_CHECKING:
from requests import sessions
__version__ = pbr.version.VersionInfo('os-service-types').version_string()
_service_type_manager = None
def get_service_types(*args, **kwargs):
def get_service_types(
session: Optional['sessions.Session'] = None,
only_remote: bool = False,
warn: bool = False,
) -> ServiceTypes:
"""Return singleton instance of the ServiceTypes object.
Parameters are all passed through to the
@@ -27,13 +37,15 @@ def get_service_types(*args, **kwargs):
.. note::
Only one singleton is kept, so if instances with different parameter
values are desired, directly calling the constructor is necessary.
Only one singleton is kept, so if instances with different parameter
values are desired, directly calling the constructor is necessary.
:returns: Singleton instance of
:class:`~os_service_types.service_types.ServiceTypes`
"""
global _service_type_manager
if not _service_type_manager:
_service_type_manager = ServiceTypes(*args, **kwargs)
_service_type_manager = ServiceTypes(
session, only_remote=only_remote, warn=warn
)
return _service_type_manager

8
os_service_types/data/__init__.py
View File

@@ -16,14 +16,18 @@
import json
import os
from os_service_types import types
__all__ = ['read_data']
DATA_DIR = os.path.dirname(__file__)
def read_data(filename):
def read_data(filename: str) -> types.ServiceTypes:
"""Return data that is shipped inside the Python package."""
filepath = os.path.join(DATA_DIR, filename)
with open(filepath) as fd:
return json.load(fd)
data: types.ServiceTypes = json.load(fd)
return data

16
os_service_types/exc.py
View File

@@ -19,13 +19,11 @@ import warnings
__all__ = ['warn', 'AliasUsageWarning']
def warn(warning, **kwargs):
"""Emit a warning that has builtin message text."""
message = textwrap.fill(textwrap.dedent(warning.details.format(**kwargs)))
warnings.warn(message, category=warning)
class _BaseWarning(Warning):
details: str
class AliasUsageWarning(Warning):
class AliasUsageWarning(_BaseWarning):
"""Use of historical service-type aliases is discouraged."""
details = """
@@ -34,9 +32,15 @@ class AliasUsageWarning(Warning):
"""
class UnofficialUsageWarning(Warning):
class UnofficialUsageWarning(_BaseWarning):
"""Use of unofficial service-types is discouraged."""
details = """
Requested service_type {given} is not a known official OpenStack project.
"""
def warn(warning: type[_BaseWarning], **kwargs: str | None) -> None:
"""Emit a warning that has builtin message text."""
message = textwrap.fill(textwrap.dedent(warning.details.format(**kwargs)))
warnings.warn(message, category=warning)

108
os_service_types/service_types.py
View File

@@ -14,9 +14,14 @@
# limitations under the License.
import copy
from typing import TYPE_CHECKING, Optional, overload
import os_service_types.data
from os_service_types import exc
from os_service_types import types
if TYPE_CHECKING:
from requests import sessions
__all__ = ['ServiceTypes']
@@ -24,9 +29,19 @@ BUILTIN_DATA = os_service_types.data.read_data('service-types.json')
SERVICE_TYPES_URL = "https://service-types.openstack.org/service-types.json"
def _normalize_type(service_type):
if service_type:
return service_type.replace('_', '-')
@overload
def _normalize_type(service_type: str) -> str: ...
@overload
def _normalize_type(service_type: None) -> None: ...
def _normalize_type(service_type: str | None) -> str | None:
if not service_type:
# TODO(stephenfin): This should be an error
return None
return service_type.replace('_', '-')
class ServiceTypes:
@@ -54,12 +69,17 @@ class ServiceTypes:
an error fetching remote data.
"""
def __init__(self, session=None, only_remote=False, warn=False):
def __init__(
self,
session: Optional['sessions.Session'] = None,
only_remote: bool = False,
warn: bool = False,
) -> None:
if not session and only_remote:
raise ValueError(
"only_remote was requested but no Session was provided."
)
self._service_types_data = BUILTIN_DATA
self._service_types_data: types.ServiceTypes = BUILTIN_DATA
self._warn = warn
if session:
try:
@@ -71,7 +91,7 @@ class ServiceTypes:
if only_remote:
raise
def _canonical_project_name(self, name):
def _canonical_project_name(self, name: str | None) -> str:
"Convert repo name to project name."
if name is None:
raise ValueError("Empty project name is not allowed")
@@ -79,52 +99,54 @@ class ServiceTypes:
return name.rpartition('/')[-1]
@property
def url(self):
def url(self) -> str:
"The URL from which the data was retrieved."
return SERVICE_TYPES_URL
@property
def version(self):
def version(self) -> str:
"The version of the data."
return self._service_types_data['version']
@property
def forward(self):
def forward(self) -> dict[str, list[str]]:
"Mapping service-type names to their aliases."
return copy.deepcopy(self._service_types_data['forward'])
@property
def reverse(self):
def reverse(self) -> dict[str, str]:
"Mapping aliases to their service-type names."
return copy.deepcopy(self._service_types_data['reverse'])
@property
def services(self):
def services(self) -> list[types.Service]:
"Full service-type data listing."
return copy.deepcopy(self._service_types_data['services'])
@property
def all_types_by_service_type(self):
def all_types_by_service_type(self) -> dict[str, list[str]]:
"Mapping of official service type to official type and aliases."
return copy.deepcopy(
self._service_types_data['all_types_by_service_type']
)
@property
def primary_service_by_project(self):
def primary_service_by_project(self) -> dict[str, types.Service]:
"Mapping of project name to the primary associated service."
return copy.deepcopy(
self._service_types_data['primary_service_by_project']
)
@property
def service_types_by_project(self):
def service_types_by_project(self) -> dict[str, list[str]]:
"Mapping of project name to a list of all associated service-types."
return copy.deepcopy(
self._service_types_data['service_types_by_project']
)
def get_official_service_data(self, service_type):
def get_official_service_data(
self, service_type: str
) -> types.Service | None:
"""Get the service data for an official service_type.
:param str service_type: The official service-type to get data for.
@@ -136,18 +158,18 @@ class ServiceTypes:
return service
return None
def get_service_data(self, service_type):
def get_service_data(self, service_type: str) -> types.Service | None:
"""Get the service data for a given service_type.
:param str service_type: The service-type or alias to get data for.
:returns dict: Service data for the service or None if not found.
"""
service_type = self.get_service_type(service_type)
if not service_type:
official_service_type = self.get_service_type(service_type)
if not official_service_type:
return None
return self.get_official_service_data(service_type)
return self.get_official_service_data(official_service_type)
def is_official(self, service_type):
def is_official(self, service_type: str) -> bool:
"""Is the given service-type an official service-type?
:param str service_type: The service-type to test.
@@ -155,7 +177,7 @@ class ServiceTypes:
"""
return self.get_official_service_data(service_type) is not None
def is_alias(self, service_type):
def is_alias(self, service_type: str) -> bool:
"""Is the given service-type an alias?
:param str service_type: The service-type to test.
@@ -164,7 +186,7 @@ class ServiceTypes:
service_type = _normalize_type(service_type)
return service_type in self._service_types_data['reverse']
def is_known(self, service_type):
def is_known(self, service_type: str) -> bool:
"""Is the given service-type an official type or an alias?
:param str service_type: The service-type to test.
@@ -172,7 +194,7 @@ class ServiceTypes:
"""
return self.is_official(service_type) or self.is_alias(service_type)
def is_match(self, requested, found):
def is_match(self, requested: str, found: str) -> bool:
"""Does a requested service-type match one found in the catalog?
A requested service-type matches a service-type in the catalog if
@@ -205,22 +227,24 @@ class ServiceTypes:
return False
def get_aliases(self, service_type):
def get_aliases(self, service_type: str) -> list[str]:
"""Returns the list of aliases for a given official service-type.
:param str service_type: An official service-type.
:returns list: List of aliases, or empty list if there are none.
:param service_type: An official service-type.
:returns: List of aliases, or empty list if there are none.
"""
service_type = _normalize_type(service_type)
return self._service_types_data['forward'].get(service_type, [])
def get_service_type(self, service_type, permissive=False):
def get_service_type(
self, service_type: str, permissive: bool = False
) -> str | None:
"""Given a possible service_type, return the official type.
:param str service_type: A potential service-type.
:param bool permissive:
:param service_type: A potential service-type.
:param permissive:
Return the original type if the given service_type is not found.
:returns str: The official service-type, or None if there is no match.
:returns: The official service-type, or None if there is no match.
"""
service_type = _normalize_type(service_type)
if self.is_official(service_type):
@@ -236,7 +260,7 @@ class ServiceTypes:
)
return official
def get_all_types(self, service_type):
def get_all_types(self, service_type: str) -> list[str]:
"""Get a list of official types and all known aliases.
:param str service_type: The service-type or alias to get data for.
@@ -245,15 +269,15 @@ class ServiceTypes:
service_type = _normalize_type(service_type)
if not self.is_known(service_type):
return [service_type]
return self.all_types_by_service_type[
self.get_service_type(service_type)
]
official_service_type = self.get_service_type(service_type)
assert official_service_type is not None
return self.all_types_by_service_type[official_service_type]
def get_project_name(self, service_type):
def get_project_name(self, service_type: str) -> str | None:
"""Return the OpenStack project name for a given service_type.
:param str service_type: An official service-type or alias.
:returns str: The OpenStack project name or None if there is no match.
:returns: The OpenStack project name or None if there is no match.
"""
service_type = _normalize_type(service_type)
service = self.get_service_data(service_type)
@@ -261,7 +285,9 @@ class ServiceTypes:
return service['project']
return None
def get_service_data_for_project(self, project_name):
def get_service_data_for_project(
self, project_name: str
) -> types.Service | None:
"""Return the service information associated with a project.
:param name: A repository or project name in the form
@@ -273,7 +299,9 @@ class ServiceTypes:
project_name = self._canonical_project_name(project_name)
return self.primary_service_by_project.get(project_name)
def get_all_service_data_for_project(self, project_name):
def get_all_service_data_for_project(
self, project_name: str
) -> list[types.Service]:
"""Return the information for every service associated with a project.
:param name: A repository or project name in the form
@@ -286,5 +314,7 @@ class ServiceTypes:
for service_type in self.service_types_by_project.get(
project_name, []
):
data.append(self.get_service_data(service_type))
service = self.get_service_data(service_type)
if service:
data.append(service)
return data

2
os_service_types/tests/base.py
View File

@@ -39,7 +39,7 @@ class TestCase(base.BaseTestCase):
self.remote_content['version'] = self.remote_version
class ServiceDataMixin:
class _BaseServiceDataTest(TestCase):
scenarios = [
(
'compute',

2
os_service_types/tests/test_builtin.py
View File

@@ -26,7 +26,7 @@ import os_service_types
from os_service_types.tests import base
class TestBuiltin(base.TestCase, base.ServiceDataMixin):
class TestBuiltin(base._BaseServiceDataTest):
def setUp(self):
super().setUp()
# Make an object with no network access

2
os_service_types/tests/test_remote.py
View File

@@ -30,7 +30,7 @@ import os_service_types.service_types
from os_service_types.tests import base
class TestRemote(base.TestCase, base.ServiceDataMixin):
class TestRemote(base._BaseServiceDataTest):
def setUp(self):
super().setUp()
# Set up a requests_mock fixture for all HTTP traffic

34
os_service_types/types.py Normal file
View File

@@ -0,0 +1,34 @@
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TypedDict
from typing_extensions import NotRequired, Required
class Service(TypedDict, total=False):
aliases: NotRequired[list[str]]
api_reference: Required[str]
project: Required[str]
service_type: Required[str]
class ServiceTypes(TypedDict):
all_types_by_service_type: Required[dict[str, list[str]]]
forward: Required[dict[str, list[str]]]
primary_service_by_project: Required[dict[str, Service]]
reverse: Required[dict[str, str]]
service_types_by_project: Required[dict[str, list[str]]]
services: Required[list[Service]]
sha: Required[str]
version: Required[str]

20
pyproject.toml
View File

@@ -1,3 +1,23 @@
[tool.mypy]
python_version = "3.10"
show_column_numbers = true
show_error_context = true
strict = true
extra_checks = true
# keep this in-sync with 'mypy.exclude' in '.pre-commit-config.yaml'
exclude = '''
(?x)(
doc
| releasenotes
)
'''
[[tool.mypy.overrides]]
module = ["os_service_types.tests.*"]
disallow_subclassing_any = false
disallow_untyped_calls = false
disallow_untyped_defs = false
[tool.ruff]
line-length = 79
target-version = "py310"

3
tox.ini
View File

@@ -54,3 +54,6 @@ commands = oslo_debug_helper {posargs}
show-source = true
builtins = _
exclude = .venv,.git,.tox,dist,doc,*lib/python*,*egg,build
[hacking]
import_exceptions = typing,typing_extensions