Commit 7067f170 authored by Felix Delattre's avatar Felix Delattre
Browse files

Introduced argument factory

parent 4f3ab5b6
Pipeline #21666 passed with stage
in 1 minute and 12 seconds
*.pyc
*.log
*.pkl
*.egg-info
Pipfile
Pipfile.lock
.pre-commit-config.yaml
__pycache__
.cache
build
dist
env
venv
.idea
#!/usr/bin/env python3
# Copyright (C) 2020-2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
...@@ -17,30 +17,26 @@ ...@@ -17,30 +17,26 @@
# along with this program. If not, see http://www.gnu.org/licenses/. # along with this program. If not, see http://www.gnu.org/licenses/.
import re import re
from argparse import ArgumentParser
def format_argument(argument):
"""Format arguments the same way as the `argparse.ArgumentParser` does."""
# strip possible preceding dashes
argument = re.sub(r"^-+", "", argument)
# replace intermediate dashes with underscore from argparse import ArgumentParser
return re.sub(r"-", "_", argument)
class BuildingIDArgument: class ArgumentBase:
def __init__(self): def __init__(self):
self.argument = "--building-id" self.argument = None
self.rabotnik_message = "building" self.rabotnik_message = None
self.help_text = None
def add_argument_to_parser(self, parser: ArgumentParser): def add_argument_to_parser(self, parser: ArgumentParser):
"""Attach this instance to a `parser`.""" """Attach this instance to a `parser`."""
parser.add_argument(
self.argument, type=str, help="Single `building_id` as integer", required=True
)
def process_arguments(self, arguments: dict): parser.add_argument(self.argument, type=str, help=self.help_text)
arg = format_argument(self.argument)
return {arg: arguments[arg]} def get_formatted_argument(self):
"""Format arguments the same way as the `argparse.ArgumentParser` does."""
# strip possible preceding dashes
formatted_argument = re.sub(r"^-+", "", self.argument)
# replace intermediate dashes with underscore
return re.sub(r"-", "_", formatted_argument)
#!/usr/bin/env python3
# Copyright (c) 2020-2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
import logging
from argparse import ArgumentParser
from .argument_base import ArgumentBase
from .arguments.building_id import ArgumentBuildingId
logger = logging.getLogger(__name__)
class ArgumentFactory:
def __init__(self):
"""Create and import argument classes based on a given selector.
Args:
selector (str): The name to find the associated argument handler
"""
self.arguments = {}
self.add_argument(ArgumentBuildingId())
def add_to_parser(self, parser: ArgumentParser):
group = parser.add_argument_group("rabotnik arguments").add_mutually_exclusive_group(
required=True
)
for argument_object in self.arguments.values():
argument_object.add_argument_to_parser(group)
def add_argument(self, argument_object: ArgumentBase):
argument = argument_object.get_formatted_argument()
self.arguments[argument] = argument_object
...@@ -15,12 +15,3 @@ ...@@ -15,12 +15,3 @@
# #
# You should have received a copy of the GNU Affero General Public License # You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/. # along with this program. If not, see http://www.gnu.org/licenses/.
from rcli.building_id_argument import format_argument
def test_format_argument():
assert format_argument("--x") == "x"
assert format_argument("--x-x") == "x_x"
assert format_argument("x-x") == "x_x"
assert format_argument("x_x") == "x_x"
#!/usr/bin/env python3
# Copyright (C) 2020-2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
from ..argument_base import ArgumentBase
import logging
logger = logging.getLogger(__name__)
class ArgumentBuildingId(ArgumentBase):
def __init__(self):
self.argument = "--building-id"
self.rabotnik_message = "building"
self.help_text = "Single `building_id` as integer"
async def execute(self, message_bus, argument, value):
logger.info(f"Dispatching `{argument}` with `{value}` on `{self.rabotnik_message}`.")
await message_bus.send(message=self.rabotnik_message, payload={argument: value})
...@@ -23,7 +23,7 @@ import logging ...@@ -23,7 +23,7 @@ import logging
from rabotnik.bus import MessageBus, PASSWORD, USERNAME, URL from rabotnik.bus import MessageBus, PASSWORD, USERNAME, URL
from .building_id_argument import BuildingIDArgument from .argument_factory import ArgumentFactory
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
...@@ -36,17 +36,17 @@ class RCLI: ...@@ -36,17 +36,17 @@ class RCLI:
Args: Args:
parser: Creates and parses command line arguments. parser: Creates and parses command line arguments.
Attributes: Attributes:
argument: A `BuildingIDArgument` that is added the `parser`. arguments: The arguments that are made available in the rcli.
""" """
def __init__(self, parser: argparse.ArgumentParser): def __init__(self, parser: argparse.ArgumentParser):
self.parser = parser self.parser = parser
self.argument = BuildingIDArgument() self.arguments = ArgumentFactory()
self.message_bus = None self.message_bus = None
def setup_parser(self): def setup_parser(self):
self.argument.add_argument_to_parser(self.parser) self.arguments.add_to_parser(self.parser)
def set_message_bus(self, message_bus: MessageBus): def set_message_bus(self, message_bus: MessageBus):
self.message_bus = message_bus self.message_bus = message_bus
...@@ -54,16 +54,15 @@ class RCLI: ...@@ -54,16 +54,15 @@ class RCLI:
async def execute(self, args): async def execute(self, args):
await self.ensure_message_bus_connected() await self.ensure_message_bus_connected()
try: # Find the argument to be put on the message bus
payload = self.argument.process_arguments(args) for key, value in args.items():
except KeyError as ex: if key in self.arguments.arguments:
logger.debug(f"Argument {ex} not provided") rabotnik_argument = key
return rabotnik_value = value
message = self.argument.rabotnik_message await self.arguments.arguments[rabotnik_argument].execute(
self.message_bus, rabotnik_argument, rabotnik_value
logger.info(f"Dispatching {payload} on {message}") )
await self.message_bus.send(message=message, payload=payload)
async def ensure_message_bus_connected(self): async def ensure_message_bus_connected(self):
if not self.message_bus: if not self.message_bus:
......
#!/usr/bin/env python3
# Copyright (C) 2020-2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
from rcli.argument_base import ArgumentBase
def test_get_format_argument():
class ArgumentTest1(ArgumentBase):
def __init__(self):
self.argument = "--x"
test_argument = ArgumentTest1()
assert test_argument.get_formatted_argument() == "x"
class ArgumentTest2(ArgumentBase):
def __init__(self):
self.argument = "--x-x"
test_argument = ArgumentTest2()
assert test_argument.get_formatted_argument() == "x_x"
class ArgumentTest3(ArgumentBase):
def __init__(self):
self.argument = "x-x"
test_argument = ArgumentTest3()
assert test_argument.get_formatted_argument() == "x_x"
class ArgumentTest4(ArgumentBase):
def __init__(self):
self.argument = "x_x"
test_argument = ArgumentTest4()
assert test_argument.get_formatted_argument() == "x_x"
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment