Skip to content

Commit

Permalink
Refactor for Python 2.7
Browse files Browse the repository at this point in the history 
Convert type hints to comments and remove `/` and `*` from functions
arguments.
  • Loading branch information
@lel-amri
lel-amri committed Feb 12, 2023
1 parent d206464 commit a960d79
Showing 1 changed file with 114 additions and 112 deletions.
226 changes: 114 additions & 112 deletions plugins/modules/docker_compose_v2.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -356,9 +356,13 @@
'''


from typing import List, Optional, Tuple, Union, Literal, Final, FrozenSet
import sys
if sys.version_info == 3:
from typing import List, Optional, Tuple, Union, Literal, Final, FrozenSet, Text, Dict, Type, NamedTuple, TYPE_CHECKING, Any
else:
TYPE_CHECKING = False
import re
from collections import defaultdict
from collections import defaultdict, namedtuple
import enum
from dataclasses import dataclass
from ansible.module_utils.basic import AnsibleModule, missing_required_lib
Expand All @@ -368,7 +372,7 @@
from ansible.module_utils.common.yaml import HAS_YAML, yaml_dump


STATUS_DONE: Final[FrozenSet[str]] = frozenset({
STATUS_DONE = frozenset({
'Started',
'Healthy',
'Exited',
Expand All @@ -380,10 +384,10 @@
'Removed',
# An extra, specific to containers
'Recreated',
})
}) # type: Final[FrozenSet[Text]]


STATUS_WORKING: Final[FrozenSet[str]] = frozenset({
STATUS_WORKING = frozenset({
'Creating',
'Starting',
'Waiting',
Expand All @@ -393,23 +397,23 @@
'Removing',
# An extra, specific to containers
'Recreate',
})
}) # type: Final[FrozenSet[Text]]


STATUS_ERROR: Final[FrozenSet[str]] = frozenset({
STATUS_ERROR = frozenset({
'Error',
})
}) # type: Final[FrozenSet[Text]]


STATUS_THAT_CAUSE_A_CHANGE = {
STATUS_THAT_CAUSE_A_CHANGE = FrozenSet({
'Started',
'Exited',
'Restarted',
'Created',
'Stopped',
'Killed',
'Removed',
}
}) # type: Final[FrozenSet[Text]]


STATUS_DOCKERCOMPOSE_TO_COMMUNITYDOCKER = {
Expand All @@ -423,17 +427,18 @@
'Killed': 'killed',
'Removed': 'removed',
'Recreated': 'recreated',
}
} # type: Final[Dict[Text, Text]]


class ResourceType(enum.Enum):
NETWORK = enum.auto()
IMAGE = enum.auto()
VOLUME = enum.auto()
CONTAINER = enum.auto()
class ResourceType(object):
NETWORK = object()
IMAGE = object()
VOLUME = object()
CONTAINER = object()

@classmethod
def from_docker_compose_event(cls, resource_type: str, /) -> "ResourceType":
def from_docker_compose_event(cls, resource_type):
# type: (Type[ResourceType], Text) -> Any
return {
"Network": cls.NETWORK,
"Image": cls.IMAGE,
Expand All @@ -442,11 +447,13 @@ def from_docker_compose_event(cls, resource_type: str, /) -> "ResourceType":
}[resource_type]


@dataclass
class ResourceEvent(object):
resource_type: ResourceType
resource_id: str
status: str
if TYPE_CHECKING:
EVENT = Tuple[Any, Text, Text]

ResourceEvent = namedtuple(
'ResourceEvent',
['resource_type', 'resource_id', 'status']
)


_re_resource_event = re.compile(r'^(?P<resource_type>Network|Image|Volume|Container) (?P<resource_id>.+) (?P<status>{:s})'.format("|".join(STATUS_DONE | STATUS_WORKING | STATUS_ERROR)))
Expand All @@ -456,35 +463,36 @@ class ResourceEvent(object):


class ComposeManager(object):
def __init__(self, module: AnsibleModule, docker_host: str, /) -> None:
def __init__(self, module, docker_host):
# type: (ComposeManager, AnsibleModule, Text) -> None
self._docker_host = docker_host
self._module = module

@staticmethod
def _parse_stderr(stderr: str) -> List[ResourceEvent]:
def _parse_stderr(stderr):
# type: (Text) -> List[EVENT]
events: List[ResourceEvent] = []
for line in stderr.splitlines():
line = line.rstrip()
if ((match := _re_resource_event.match(line)) is not None):
events.append(ResourceEvent(
resource_type=ResourceType.from_docker_compose_event(match.group('resource_type')),
resource_id=match.group('resource_id'),
status=match.group('status'))
)
ResourceType.from_docker_compose_event(match.group('resource_type')),
match.group('resource_id'),
match.group('status')
))
return events

def _run_subcommand(
self,
subcommand: List[str],
files: List[str],
/,
content: Optional[str] = None,
*,
project_name: Optional[str] = None,
project_directory: Optional[str] = None,
profiles: List[str] = [],
env_file: Optional[str],
) -> Tuple[int, str, str, List[ResourceEvent]]:
subcommand, # type: List[Text]
files, # type: List[Text]
content = None, # type: Optional[Text]
project_name = None, # type: Optional[Text]
project_directory = None, # type: Optional[Text]
profiles = [], # type: List[Text]
env_file = None, # type: Optional[Text]
):
# type: (...) -> Tuple[int, Text, Text, List[EVENT]]
command = [DOCKER_COMPOSE_EXECUTABLE, '--ansi', 'never']
for file in files:
command.extend(['-f', file])
Expand Down Expand Up @@ -520,24 +528,23 @@ def _run_subcommand(
def up(
self,
# Common arguments
files: List[str],
/,
content: Optional[str] = None,
*,
project_name: Optional[str] = None,
project_directory: Optional[str] = None,
profiles: List[str] = [],
env_file: Optional[str],
files, # type: List[Text]
content = None, # type: Optional[Text]
project_name = None, # type: Optional[Text]
project_directory = None, # type: Optional[Text]
profiles = [], # type: List[Text]
env_file = None, # type: Optional[Text]
# Specific arguments
services: List[str] = [],
no_deps: bool = False,
pull: Optional[Union[Literal['always'], Literal['missing'], Literal['never']]] = None,
build: bool = False,
force_recreate: bool = False,
no_recreate: bool = False,
remove_orphans: bool = False,
timeout: Optional[int] = None,
) -> Tuple[int, str, str, List[ResourceEvent]]:
services = [], # type: List[Text]
no_deps = False, # type: bool
pull = None, # type: Optional[Union[Literal['always'], Literal['missing'], Literal['never']]]
build = False, # type: bool
force_recreate = False, # type: bool
no_recreate = False, # type: bool
remove_orphans = False, # type: bool
timeout = None, # type: Optional[int]
):
# type: (...) -> Tuple[int, Text, Text, List[EVENT]]
subcommand = ['up', '-d']
if no_deps:
subcommand.append('--no-deps')
Expand Down Expand Up @@ -568,20 +575,19 @@ def up(
def down(
self,
# Common arguments
files: List[str],
/,
content: Optional[str] = None,
*,
project_name: Optional[str] = None,
project_directory: Optional[str] = None,
profiles: List[str] = [],
env_file: Optional[str],
files, # type: List[Text]
content = None, # type: Optional[Text]
project_name = None, # type: Optional[Text]
project_directory = None, # type: Optional[Text]
profiles = [], # type: List[Text]
env_file = None, # type: Optional[Text]
# Specific arguments
remove_orphans: bool = False,
rmi: Optional[Union[Literal['all'], Literal['local']]] = None,
volumes: bool = False,
timeout: Optional[int] = None,
) -> Tuple[int, str, str, List[ResourceEvent]]:
remove_orphans: bool = False, # type: bool
rmi = None, # type: Optional[Union[Literal['all'], Literal['local']]]
volumes = False, # type: bool
timeout = None, # type: Optional[int]
):
# type: (...) -> Tuple[int, Text, Text, List[EVENT]]
subcommand = ['down']
if remove_orphans:
subcommand.append('--remove-orphans')
Expand All @@ -604,17 +610,16 @@ def down(
def stop(
self,
# Common arguments
files: List[str],
/,
content: Optional[str] = None,
*,
project_name: Optional[str] = None,
project_directory: Optional[str] = None,
profiles: List[str] = [],
env_file: Optional[str],
files, # type: List[Text]
content = None, # type: Optional[Text]
project_name = None, # type: Optional[Text]
project_directory = None, # type: Optional[Text]
profiles = [], # type: List[Text]
env_file = None, # type: Optional[Text]
# Specific arguments
timeout: Optional[int] = None,
) -> Tuple[int, str, str, List[ResourceEvent]]:
timeout = None, # type: Optional[int]
):
# type: (...) -> Tuple[int, Text, Text, List[EVENT]]
subcommand = ['stop']
if timeout is not None:
subcommand.extend(['--timeout', '{:d}'.format(timeout)])
Expand All @@ -631,18 +636,17 @@ def stop(
def restart(
self,
# Common arguments
files: List[str],
/,
content: Optional[str] = None,
*,
project_name: Optional[str] = None,
project_directory: Optional[str] = None,
profiles: List[str] = [],
env_file: Optional[str],
files, # type: List[Text]
content = None, # type: Optional[Text]
project_name = None, # type: Optional[Text]
project_directory = None, # type: Optional[Text]
profiles = [], # type: List[Text]
env_file = None, # type: Optional[Text]
# Specific arguments
services: List[str] = [],
timeout: Optional[int] = None,
) -> Tuple[int, str, str, List[ResourceEvent]]:
services = [], # type: List[Text]
timeout = None, # type: Optional[int]
):
# type: (...) -> Tuple[int, Text, Text, List[EVENT]]
subcommand = ['restart']
if timeout is not None:
subcommand.extend(['--timeout', '{:d}'.format(timeout)])
Expand All @@ -661,19 +665,18 @@ def restart(
def build(
self,
# Common arguments
files: List[str],
/,
content: Optional[str] = None,
*,
project_name: Optional[str] = None,
project_directory: Optional[str] = None,
profiles: List[str] = [],
env_file: Optional[str],
files, # type: List[Text]
content = None, # type: Optional[Text]
project_name = None, # type: Optional[Text]
project_directory = None, # type: Optional[Text]
profiles = [], # type: List[Text]
env_file = None, # type: Optional[Text]
# Specific arguments
services: List[str] = [],
no_cache: bool = False,
pull: bool = False,
) -> Tuple[int, str, str, List[ResourceEvent]]:
services = [], # type: List[Text]
no_cache= False, # type: bool
pull= False, # type: bool
):
# type: (...) -> Tuple[int, Text, Text, List[EVENT]]
subcommand = ['build']
if no_cache:
subcommand.append('--no-cache')
Expand All @@ -694,18 +697,17 @@ def build(
def pull(
self,
# Common arguments
files: List[str],
/,
content: Optional[str] = None,
*,
project_name: Optional[str] = None,
project_directory: Optional[str] = None,
profiles: List[str] = [],
env_file: Optional[str],
files, # type: List[Text]
content = None, # type: Optional[Text]
project_name = None, # type: Optional[Text]
project_directory = None, # type: Optional[Text]
profiles = [], # type: List[Text]
env_file = None, # type: Optional[Text]
# Specific arguments
services: List[str] = [],
include_deps: bool = False,
) -> Tuple[int, str, str, List[ResourceEvent]]:
services = [], # type: List[Text]
include_deps = False, # type: bool
):
# type: (...) -> Tuple[int, Text, Text, List[EVENT]]
subcommand = ['pull']
if include_deps:
subcommand.append('--include-deps')
Expand Down

0 comments on commit a960d79

Please sign in to comment.