x
/
ospurge
Code Issues Proposed changes

Python 2: Move type annotations to .pyi files 

Python 2 doesn't support type annotations: raise SyntaxError. Move
them to separated .pyi files to add Python 2 support without loosing
annotations.

Keep inline comments like "# type: Optional[str]".

Related-Bug: 1726399
Change-Id: Ib1a837e88cf76908f1007b3881703ffb433e6c2f
This commit is contained in:
Victor Stinner 2017-10-23 16:15:22 +02:00
parent 44cab87d5f
commit b73c938a93
17 changed files with 539 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
ospurge/main.py
View File

@ -22,14 +22,13 @@ import os_client_config
import shade
from ospurge import exceptions
from ospurge.resources.base import ServiceResource
from ospurge import utils
if typing.TYPE_CHECKING: # pragma: no cover
from typing import Optional # noqa: F401
def configure_logging(verbose: bool) -> None:
def configure_logging(verbose):
log_level = logging.INFO if verbose else logging.WARNING
logging.basicConfig(
format='%(levelname)s:%(name)s:%(asctime)s:%(message)s',
@ -39,7 +38,7 @@ def configure_logging(verbose: bool) -> None:
'requests.packages.urllib3.connectionpool').setLevel(logging.WARNING)
def create_argument_parser() -> argparse.ArgumentParser:
def create_argument_parser():
parser = argparse.ArgumentParser(
description="Purge resources from an Openstack project."
)
@ -84,7 +83,7 @@ def create_argument_parser() -> argparse.ArgumentParser:
class CredentialsManager(object):
def __init__(self, options: argparse.Namespace) -> None:
def __init__(self, options):
self.options = options
self.revoke_role_after_purge = False
@ -128,7 +127,7 @@ class CredentialsManager(object):
or auth_args.get('project_id')
)
def ensure_role_on_project(self) -> None:
def ensure_role_on_project(self):
if self.operator_cloud and self.operator_cloud.grant_role(
self.options.admin_role_name,
project=self.options.purge_project, user=self.user_id
@ -139,7 +138,7 @@ class CredentialsManager(object):
)
self.revoke_role_after_purge = True
def revoke_role_on_project(self) -> None:
def revoke_role_on_project(self):
self.operator_cloud.revoke_role(
self.options.admin_role_name, user=self.user_id,
project=self.options.purge_project)
@ -148,22 +147,19 @@ class CredentialsManager(object):
self.user_id, self.options.purge_project
)
def ensure_enabled_project(self) -> None:
def ensure_enabled_project(self):
if self.operator_cloud and self.disable_project_after_purge:
self.operator_cloud.update_project(self.project_id, enabled=True)
logging.warning("Project '%s' was disabled before purge and it is "
"now enabled", self.options.purge_project)
def disable_project(self) -> None:
def disable_project(self):
self.operator_cloud.update_project(self.project_id, enabled=False)
logging.warning("Project '%s' was disabled before purge and it is "
"now also disabled", self.options.purge_project)
def runner(
resource_mngr: ServiceResource, options: argparse.Namespace,
exit: threading.Event
) -> None:
def runner(resource_mngr, options, exit):
try:
if not (options.dry_run or options.resource):
@ -216,7 +212,7 @@ def runner(
@utils.monkeypatch_oscc_logging_warning
def main() -> None:
def main():
parser = create_argument_parser()
cloud_config = os_client_config.OpenStackConfig()
@ -242,7 +238,7 @@ def main() -> None:
# Dummy function to work around `ThreadPoolExecutor.map()` not accepting
# a callable with arguments.
def partial_runner(resource_manager: ServiceResource) -> None:
def partial_runner(resource_manager):
runner(resource_manager, options=options,
exit=exit) # pragma: no cover

56
ospurge/main.pyi Normal file
View File

@ -0,0 +1,56 @@
#!/usr/bin/env python3
# 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.
import argparse
import threading
import typing
from typing import Optional # noqa: F401
from ospurge.resources.base import ServiceResource
from ospurge import utils
def configure_logging(verbose: bool) -> None:
...
def create_argument_parser() -> argparse.ArgumentParser:
...
class CredentialsManager(object):
def __init__(self, options: argparse.Namespace) -> None:
...
def ensure_role_on_project(self) -> None:
...
def revoke_role_on_project(self) -> None:
...
def ensure_enabled_project(self) -> None:
...
def disable_project(self) -> None:
...
def runner(
resource_mngr: ServiceResource, options: argparse.Namespace,
exit: threading.Event
) -> None:
...
@utils.monkeypatch_oscc_logging_warning
def main() -> None:
...

36
ospurge/resources/base.py
View File

@ -12,12 +12,7 @@
import abc
import collections
import logging
import threading
import time
from typing import Any
from typing import Dict
from typing import Iterable
from typing import Optional
from typing import TYPE_CHECKING
try:
@ -29,16 +24,12 @@ from ospurge import exceptions
if TYPE_CHECKING: # pragma: no cover
import argparse # noqa: F401
from ospurge.main import CredentialsManager # noqa: F401
import shade # noqa: F401
from typing import Optional # noqa: F401
class MatchSignaturesMeta(type):
def __init__(
self, clsname: str, bases: Optional[Any],
clsdict: Optional[Dict]
) -> None:
def __init__(self, clsname, bases, clsdict):
super().__init__(clsname, bases, clsdict)
sup = super(self, self) # type: ignore # See python/mypy #857
for name, value in clsdict.items():
@ -57,10 +48,7 @@ class MatchSignaturesMeta(type):
class OrderedMeta(type):
def __new__(
cls, clsname: str, bases: Optional[Any],
clsdict: Optional[Dict]
) -> type:
def __new__(cls, clsname, bases, clsdict):
ordered_methods = cls.ordered_methods
allowed_next_methods = list(ordered_methods)
for name, value in clsdict.items():
@ -83,7 +71,7 @@ class OrderedMeta(type):
return super().__new__(cls, clsname, bases, dict(clsdict))
@classmethod
def __prepare__(cls, clsname: str, bases: Optional[Any]) -> Dict:
def __prepare__(cls, clsname, bases):
return collections.OrderedDict()
@ -93,7 +81,7 @@ class CodingStyleMixin(OrderedMeta, MatchSignaturesMeta, abc.ABCMeta):
class BaseServiceResource(object):
def __init__(self) -> None:
def __init__(self):
self.cleanup_project_id = None # type: Optional[str]
self.cloud = None # type: Optional[shade.OpenStackCloud]
self.options = None # type: Optional[argparse.Namespace]
@ -102,7 +90,7 @@ class BaseServiceResource(object):
class ServiceResource(BaseServiceResource, metaclass=CodingStyleMixin):
ORDER = None # type: int
def __init__(self, creds_manager: 'CredentialsManager') -> None:
def __init__(self, creds_manager):
super().__init__()
if self.ORDER is None:
raise ValueError(
@ -115,17 +103,17 @@ class ServiceResource(BaseServiceResource, metaclass=CodingStyleMixin):
self.options = creds_manager.options
@classmethod
def order(cls) -> int:
def order(cls):
return cls.ORDER
def check_prerequisite(self) -> bool:
def check_prerequisite(self):
return True
@abc.abstractmethod
def list(self) -> Iterable:
def list(self):
raise NotImplementedError
def should_delete(self, resource: Dict[str, Any]) -> bool:
def should_delete(self, resource):
project_id = resource.get('project_id', resource.get('tenant_id'))
if project_id:
return project_id == self.cleanup_project_id
@ -137,15 +125,15 @@ class ServiceResource(BaseServiceResource, metaclass=CodingStyleMixin):
return True
@abc.abstractmethod
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
raise NotImplementedError
@staticmethod
@abc.abstractmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
raise NotImplementedError
def wait_for_check_prerequisite(self, exit: threading.Event) -> None:
def wait_for_check_prerequisite(self, exit):
timeout = time.time() + 120
sleep = 2
while time.time() < timeout:

79
ospurge/resources/base.pyi Normal file
View File

@ -0,0 +1,79 @@
# 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.
import abc
import threading
from typing import Any
from typing import Dict
from typing import Iterable
from typing import Optional
from ospurge.main import CredentialsManager # noqa: F401
class MatchSignaturesMeta(type):
def __init__(
self, clsname: str, bases: Optional[Any],
clsdict: Optional[Dict]
) -> None:
...
class OrderedMeta(type):
def __new__(
cls, clsname: str, bases: Optional[Any],
clsdict: Optional[Dict]
) -> type:
...
@classmethod
def __prepare__(cls, clsname: str, bases: Optional[Any]) -> Dict:
...
class CodingStyleMixin(OrderedMeta, MatchSignaturesMeta, abc.ABCMeta):
...
class BaseServiceResource(object):
def __init__(self) -> None:
...
class ServiceResource(BaseServiceResource, metaclass=CodingStyleMixin):
def __init__(self, creds_manager: 'CredentialsManager') -> None:
...
@classmethod
def order(cls) -> int:
...
def check_prerequisite(self) -> bool:
...
@abc.abstractmethod
def list(self) -> Iterable:
...
def should_delete(self, resource: Dict[str, Any]) -> bool:
...
@abc.abstractmethod
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
@abc.abstractmethod
def to_str(resource: Dict[str, Any]) -> str:
...
def wait_for_check_prerequisite(self, exit: threading.Event) -> None:
...

26
ospurge/resources/cinder.py
View File

@ -9,24 +9,20 @@
# 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 Any
from typing import Dict
from typing import Iterable
from ospurge.resources import base
class Backups(base.ServiceResource):
ORDER = 33
def list(self) -> Iterable:
def list(self):
return self.cloud.list_volume_backups()
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_volume_backup(resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Volume Backup (id='{}', name='{}'".format(
resource['id'], resource['name'])
@ -34,14 +30,14 @@ class Backups(base.ServiceResource):
class Snapshots(base.ServiceResource):
ORDER = 36
def list(self) -> Iterable:
def list(self):
return self.cloud.list_volume_snapshots()
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_volume_snapshot(resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Volume Snapshot (id='{}', name='{}')".format(
resource['id'], resource['name'])
@ -49,21 +45,21 @@ class Snapshots(base.ServiceResource):
class Volumes(base.ServiceResource):
ORDER = 65
def check_prerequisite(self) -> bool:
def check_prerequisite(self):
return (self.cloud.list_volume_snapshots() == [] and
self.cloud.list_servers() == [])
def list(self) -> Iterable:
def list(self):
return self.cloud.list_volumes()
def should_delete(self, resource: Dict[str, Any]) -> bool:
def should_delete(self, resource):
attr = 'os-vol-tenant-attr:tenant_id'
return resource[attr] == self.cleanup_project_id
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_volume(resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Volume (id='{}', name='{}')".format(
resource['id'], resource['name'])

58
ospurge/resources/cinder.pyi Normal file
View File

@ -0,0 +1,58 @@
# 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 Any
from typing import Dict
from typing import Iterable
from ospurge.resources import base
class Backups(base.ServiceResource):
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...
class Snapshots(base.ServiceResource):
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...
class Volumes(base.ServiceResource):
def check_prerequisite(self) -> bool:
...
def list(self) -> Iterable:
...
def should_delete(self, resource: Dict[str, Any]) -> bool:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...

14
ospurge/resources/glance.py
View File

@ -9,16 +9,12 @@
# 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 Any
from typing import Dict
from typing import Iterable
from ospurge.resources import base
from ospurge.resources.base import BaseServiceResource
class ListImagesMixin(BaseServiceResource):
def list_images_by_owner(self) -> Iterable[Dict[str, Any]]:
def list_images_by_owner(self):
images = []
for image in self.cloud.list_images():
if image['owner'] != self.cleanup_project_id:
@ -38,16 +34,16 @@ class ListImagesMixin(BaseServiceResource):
class Images(base.ServiceResource, ListImagesMixin):
ORDER = 53
def list(self) -> Iterable:
def list(self):
return self.list_images_by_owner()
def should_delete(self, resource: Dict[str, Any]) -> bool:
def should_delete(self, resource):
return resource['owner'] == self.cleanup_project_id
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_image(resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Image (id='{}', name='{}')".format(
resource['id'], resource['name'])

37
ospurge/resources/glance.pyi Normal file
View File

@ -0,0 +1,37 @@
# 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 Any
from typing import Dict
from typing import Iterable
from ospurge.resources import base
from ospurge.resources.base import BaseServiceResource
class ListImagesMixin(BaseServiceResource):
def list_images_by_owner(self) -> Iterable[Dict[str, Any]]:
...
class Images(base.ServiceResource, ListImagesMixin):
def list(self) -> Iterable:
...
def should_delete(self, resource: Dict[str, Any]) -> bool:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...

48
ospurge/resources/neutron.py
View File

@ -9,37 +9,33 @@
# 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 Any
from typing import Dict
from typing import Iterable
from ospurge.resources import base
class FloatingIPs(base.ServiceResource):
ORDER = 25
def check_prerequisite(self) -> bool:
def check_prerequisite(self):
# We can't delete a FIP if it's attached
return self.cloud.list_servers() == []
def list(self) -> Iterable:
def list(self):
return self.cloud.search_floating_ips(filters={
'tenant_id': self.cleanup_project_id
})
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_floating_ip(resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Floating IP (id='{}')".format(resource['id'])
class RouterInterfaces(base.ServiceResource):
ORDER = 42
def check_prerequisite(self) -> bool:
def check_prerequisite(self):
return (
self.cloud.list_servers() == [] and
self.cloud.search_floating_ips(
@ -47,18 +43,18 @@ class RouterInterfaces(base.ServiceResource):
) == []
)
def list(self) -> Iterable:
def list(self):
return self.cloud.list_ports(
filters={'device_owner': 'network:router_interface',
'tenant_id': self.cleanup_project_id}
)
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.remove_router_interface({'id': resource['device_id']},
port_id=resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Router Interface (id='{}', router_id='{}')".format(
resource['id'], resource['device_id'])
@ -66,20 +62,20 @@ class RouterInterfaces(base.ServiceResource):
class Routers(base.ServiceResource):
ORDER = 44
def check_prerequisite(self) -> bool:
def check_prerequisite(self):
return self.cloud.list_ports(
filters={'device_owner': 'network:router_interface',
'tenant_id': self.cleanup_project_id}
) == []
def list(self) -> Iterable:
def list(self):
return self.cloud.list_routers()
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_router(resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Router (id='{}', name='{}')".format(
resource['id'], resource['name'])
@ -87,18 +83,18 @@ class Routers(base.ServiceResource):
class Ports(base.ServiceResource):
ORDER = 46
def list(self) -> Iterable:
def list(self):
ports = self.cloud.list_ports(
filters={'tenant_id': self.cleanup_project_id}
)
excluded = ['network:dhcp', 'network:router_interface']
return [p for p in ports if p['device_owner'] not in excluded]
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_port(resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Port (id='{}', network_id='{}, device_owner='{}')'".format(
resource['id'], resource['network_id'], resource['device_owner'])
@ -106,14 +102,14 @@ class Ports(base.ServiceResource):
class Networks(base.ServiceResource):
ORDER = 48
def check_prerequisite(self) -> bool:
def check_prerequisite(self):
ports = self.cloud.list_ports(
filters={'tenant_id': self.cleanup_project_id}
)
excluded = ['network:dhcp']
return [p for p in ports if p['device_owner'] not in excluded] == []
def list(self) -> Iterable:
def list(self):
networks = []
for network in self.cloud.list_networks(
filters={'tenant_id': self.cleanup_project_id}
@ -125,11 +121,11 @@ class Networks(base.ServiceResource):
return networks
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_network(resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Network (id='{}', name='{}')".format(
resource['id'], resource['name'])
@ -137,15 +133,15 @@ class Networks(base.ServiceResource):
class SecurityGroups(base.ServiceResource):
ORDER = 49
def list(self) -> Iterable:
def list(self):
return [sg for sg in self.cloud.list_security_groups(
filters={'tenant_id': self.cleanup_project_id})
if sg['name'] != 'default']
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_security_group(resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Security Group (id='{}', name='{}')".format(
resource['id'], resource['name'])

100
ospurge/resources/neutron.pyi Normal file
View File

@ -0,0 +1,100 @@
# 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 Any
from typing import Dict
from typing import Iterable
from ospurge.resources import base
class FloatingIPs(base.ServiceResource):
def check_prerequisite(self) -> bool:
...
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...
class RouterInterfaces(base.ServiceResource):
def check_prerequisite(self) -> bool:
...
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...
class Routers(base.ServiceResource):
def check_prerequisite(self) -> bool:
...
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...
class Ports(base.ServiceResource):
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...
class Networks(base.ServiceResource):
def check_prerequisite(self) -> bool:
...
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...
class SecurityGroups(base.ServiceResource):
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...

10
ospurge/resources/nova.py
View File

@ -9,23 +9,19 @@
# 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 Any
from typing import Dict
from typing import Iterable
from ospurge.resources import base
class Servers(base.ServiceResource):
ORDER = 15
def list(self) -> Iterable:
def list(self):
return self.cloud.list_servers()
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_server(resource['id'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "VM (id='{}', name='{}')".format(
resource['id'], resource['name'])

28
ospurge/resources/nova.pyi Normal file
View File

@ -0,0 +1,28 @@
# 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 Any
from typing import Dict
from typing import Iterable
from ospurge.resources import base
class Servers(base.ServiceResource):
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...

23
ospurge/resources/swift.py
View File

@ -9,18 +9,13 @@
# 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 Any
from typing import Dict
from typing import Iterable
from typing import Iterator
from ospurge.resources import base
from ospurge.resources.base import BaseServiceResource
from ospurge.resources import glance
class ListObjectsMixin(BaseServiceResource):
def list_objects(self) -> Iterator[Dict[str, Any]]:
def list_objects(self):
for container in self.cloud.list_containers():
for obj in self.cloud.list_objects(container['name']):
obj['container_name'] = container['name']
@ -30,18 +25,18 @@ class ListObjectsMixin(BaseServiceResource):
class Objects(base.ServiceResource, glance.ListImagesMixin, ListObjectsMixin):
ORDER = 73
def check_prerequisite(self) -> bool:
def check_prerequisite(self):
return (self.list_images_by_owner() == [] and
self.cloud.list_volume_backups() == [])
def list(self) -> Iterable:
def list(self):
yield from self.list_objects()
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_object(resource['container_name'], resource['name'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Object '{}' from Container '{}'".format(
resource['name'], resource['container_name'])
@ -49,15 +44,15 @@ class Objects(base.ServiceResource, glance.ListImagesMixin, ListObjectsMixin):
class Containers(base.ServiceResource, ListObjectsMixin):
ORDER = 75
def check_prerequisite(self) -> bool:
def check_prerequisite(self):
return list(self.list_objects()) == []
def list(self) -> Iterable:
def list(self):
return self.cloud.list_containers()
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
self.cloud.delete_container(resource['name'])
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
return "Container (name='{}')".format(resource['name'])

54
ospurge/resources/swift.pyi Normal file
View File

@ -0,0 +1,54 @@
# 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 Any
from typing import Dict
from typing import Iterable
from typing import Iterator
from ospurge.resources import base
from ospurge.resources.base import BaseServiceResource
from ospurge.resources import glance
class ListObjectsMixin(BaseServiceResource):
def list_objects(self) -> Iterator[Dict[str, Any]]:
...
class Objects(base.ServiceResource, glance.ListImagesMixin, ListObjectsMixin):
def check_prerequisite(self) -> bool:
...
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...
class Containers(base.ServiceResource, ListObjectsMixin):
def check_prerequisite(self) -> bool:
...
def list(self) -> Iterable:
...
def delete(self, resource: Dict[str, Any]) -> None:
...
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
...

9
ospurge/tests/resources/test_base.py
View File

@ -10,9 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import time
from typing import Any
from typing import Dict
from typing import Iterable
from ospurge import exceptions
from ospurge.resources import base
@ -141,14 +138,14 @@ class TestOrderedMeta(unittest.TestCase):
class TestServiceResource(unittest.TestCase):
def test_init_without_order_attr(self):
class Foo5(base.ServiceResource):
def list(self) -> Iterable:
def list(self):
pass
def delete(self, resource: Dict[str, Any]) -> None:
def delete(self, resource):
pass
@staticmethod
def to_str(resource: Dict[str, Any]) -> str:
def to_str(resource):
pass
self.assertRaisesRegex(ValueError, 'Class .*ORDER.*',

26
ospurge/utils.py
View File

@ -17,19 +17,10 @@ import os
import pkgutil
import re
from typing import Any
from typing import Callable
from typing import cast
from typing import Dict
from typing import Iterable
from typing import List
from typing import Optional
from typing import TypeVar
from ospurge.resources import base
def get_resource_classes(resources: Optional[Iterable[str]]=None) -> List:
def get_resource_classes(resources=None):
"""
Import all the modules in the `resources` package and return all the
subclasses of the `ServiceResource` ABC that match the `resources` arg.
@ -58,10 +49,7 @@ def get_resource_classes(resources: Optional[Iterable[str]]=None) -> List:
return [c for c in all_classes if regex.match(c.__name__)]
F = TypeVar('F', bound=Callable[..., Any])
def monkeypatch_oscc_logging_warning(f: F) -> F:
def monkeypatch_oscc_logging_warning(f):
"""
Monkey-patch logging.warning() method to silence 'os_client_config' when
it complains that a Keystone catalog entry is not found. This warning
@ -71,29 +59,29 @@ def monkeypatch_oscc_logging_warning(f: F) -> F:
oscc_target = 'os_client_config.cloud_config'
orig_logging = logging.getLogger(oscc_target).warning
def logging_warning(msg: str, *args: Any, **kwargs: Any) -> None:
def logging_warning(msg: str, *args, **kwargs):
if 'catalog entry not found' not in msg:
orig_logging(msg, *args, **kwargs)
@functools.wraps(f)
def wrapper(*args: list, **kwargs: dict) -> Any:
def wrapper(*args: list, **kwargs):
try:
setattr(logging.getLogger(oscc_target), 'warning', logging_warning)
return f(*args, **kwargs)
finally:
setattr(logging.getLogger(oscc_target), 'warning', orig_logging)
return cast(F, wrapper)
return wrapper
def call_and_ignore_exc(exc: type, f: Callable, *args: List) -> None:
def call_and_ignore_exc(exc, f, *args):
try:
f(*args)
except exc as e:
logging.debug("The following exception was ignored: %r", e)
def replace_project_info(config: Dict, new_project_id: str) -> Dict[str, Any]:
def replace_project_info(config, new_project_id):
"""
Replace all tenant/project info in a `os_client_config` config dict with
a new project. This is used to bind/scope to another project.

45
ospurge/utils.pyi Normal file
View File

@ -0,0 +1,45 @@
# 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 Any
from typing import Callable
from typing import cast
from typing import Dict
from typing import Iterable
from typing import List
from typing import Optional
from typing import TypeVar
def get_resource_classes(resources: Optional[Iterable[str]]=None) -> List:
...
F = TypeVar('F', bound=Callable[..., Any])
def monkeypatch_oscc_logging_warning(f: F) -> F:
def logging_warning(msg: str, *args: Any, **kwargs: Any) -> None:
...
@functools.wraps(f)
def wrapper(*args: list, **kwargs: dict) -> Any:
...
return cast(F, wrapper)
def call_and_ignore_exc(exc: type, f: Callable, *args: List) -> None:
...
def replace_project_info(config: Dict, new_project_id: str) -> Dict[str, Any]:
...