[elbe-devel] [PATCH 1/2] elbepack: prjrepo: migrate to argparse
Thomas Weißschuh
thomas.weissschuh at linutronix.de
Wed Jul 24 15:09:16 CEST 2024
argparse has various advantages of optparse:
* Autogenerated command synopsis.
* Required arguments.
* Flexible argument types.
* Subparsers.
Furthermore optparse is deprecated since Python 3.2 (2011).
Replace the custom action registry with a simple dispatch table, which
is both less code and easier to understand.
Signed-off-by: Thomas Weißschuh <thomas.weissschuh at linutronix.de>
---
elbepack/commands/prjrepo.py | 78 +++++++-------
elbepack/soapclient.py | 243 +++++++++++++++++--------------------------
2 files changed, 132 insertions(+), 189 deletions(-)
diff --git a/elbepack/commands/prjrepo.py b/elbepack/commands/prjrepo.py
index 5d4b4acef52a..e63da2e6f37c 100644
--- a/elbepack/commands/prjrepo.py
+++ b/elbepack/commands/prjrepo.py
@@ -2,69 +2,71 @@
# SPDX-License-Identifier: GPL-3.0-or-later
# SPDX-FileCopyrightText: 2017 Linutronix GmbH
+import argparse
import socket
import sys
from http.client import BadStatusLine
-from optparse import OptionGroup, OptionParser
from urllib.error import URLError
from suds import WebFault
+from elbepack.cli import add_arguments_from_decorated_function
from elbepack.config import cfg
-from elbepack.soapclient import ElbeSoapClient, RepoAction
+from elbepack.soapclient import ElbeSoapClient, repo_actions
def run_command(argv):
- oparser = OptionParser(usage='usage: elbe prjrepo [options] <command>')
+ aparser = argparse.ArgumentParser(prog='elbe prjrepo')
- oparser.add_option('--host', dest='host', default=cfg['soaphost'],
- help='Ip or hostname of elbe-daemon.')
+ aparser.add_argument('--host', dest='host', default=cfg['soaphost'],
+ help='Ip or hostname of elbe-daemon.')
- oparser.add_option('--port', dest='port', default=cfg['soapport'],
- help='Port of soap itf on elbe-daemon.')
+ aparser.add_argument('--port', dest='port', default=cfg['soapport'],
+ help='Port of soap itf on elbe-daemon.')
- oparser.add_option('--pass', dest='passwd', default=cfg['elbepass'],
- help='Password (default is foo).')
+ aparser.add_argument('--pass', dest='passwd', default=cfg['elbepass'],
+ help='Password (default is foo).')
- oparser.add_option('--user', dest='user', default=cfg['elbeuser'],
- help='Username (default is root).')
+ aparser.add_argument('--user', dest='user', default=cfg['elbeuser'],
+ help='Username (default is root).')
- oparser.add_option(
+ aparser.add_argument(
'--retries',
dest='retries',
- default='10',
+ type=int, default=10,
help='How many times to retry the connection to the server before\
giving up (default is 10 times, yielding 10 seconds).')
- devel = OptionGroup(
- oparser,
+ devel = aparser.add_argument_group(
'options for elbe developers',
"Caution: Don't use these options in a productive environment")
- devel.add_option('--debug', action='store_true',
- dest='debug', default=False,
- help='Enable debug mode.')
+ devel.add_argument('--debug', action='store_true',
+ dest='debug', default=False,
+ help='Enable debug mode.')
- (opt, args) = oparser.parse_args(argv)
+ subparsers = aparser.add_subparsers(required=True)
- if not args:
- print('elbe prjrepo - no subcommand given', file=sys.stderr)
- RepoAction.print_actions()
- return
+ for action_name, do_action in repo_actions.items():
+ action_parser = subparsers.add_parser(action_name)
+ action_parser.set_defaults(func=do_action)
+ add_arguments_from_decorated_function(action_parser, do_action)
+
+ args = aparser.parse_args(argv)
# Try to connect to initvm via SOAP
try:
control = ElbeSoapClient(
- opt.host,
- opt.port,
- opt.user,
- opt.passwd,
- debug=opt.debug,
- retries=int(
- opt.retries))
+ args.host,
+ args.port,
+ args.user,
+ args.passwd,
+ debug=args.debug,
+ retries=args.retries,
+ )
except URLError:
print(
- f'Failed to connect to Soap server {opt.host}:{opt.port}\n',
+ f'Failed to connect to Soap server {args.host}:{args.port}\n',
file=sys.stderr)
print('', file=sys.stderr)
print('Check, wether the initvm is actually running.', file=sys.stderr)
@@ -72,7 +74,7 @@ def run_command(argv):
sys.exit(10)
except socket.error:
print(
- f'Failed to connect to Soap server {opt.host}:{opt.port}\n',
+ f'Failed to connect to Soap server {args.host}:{args.port}\n',
file=sys.stderr)
print('', file=sys.stderr)
print(
@@ -82,7 +84,7 @@ def run_command(argv):
sys.exit(11)
except BadStatusLine:
print(
- f'Failed to connect to Soap server {opt.host}:{opt.port}\n',
+ f'Failed to connect to Soap server {args.host}:{args.port}\n',
file=sys.stderr)
print('', file=sys.stderr)
print('Check, wether the initvm is actually running.', file=sys.stderr)
@@ -91,17 +93,9 @@ def run_command(argv):
file=sys.stderr)
sys.exit(12)
- # Check whether subcommand exists
- try:
- action = RepoAction(args[0])
- except KeyError:
- print('elbe prjrepo - unknown subcommand', file=sys.stderr)
- RepoAction.print_actions()
- sys.exit(22)
-
# Execute command
try:
- action.execute(control, opt, args[1:])
+ args.func(control, args)
except WebFault as e:
print('Server returned an error:', file=sys.stderr)
print('', file=sys.stderr)
diff --git a/elbepack/soapclient.py b/elbepack/soapclient.py
index 8d0b53c77b3c..621bfdb49539 100644
--- a/elbepack/soapclient.py
+++ b/elbepack/soapclient.py
@@ -17,6 +17,7 @@ import debian.deb822
from suds.client import Client
+from elbepack.cli import add_argument
from elbepack.config import cfg
from elbepack.version import elbe_version
@@ -122,158 +123,106 @@ class ElbeSoapClient:
part = part + 1
-class RepoAction:
- repoactiondict = {}
+ at add_argument('project_dir')
+def _list_packages(client, args):
+ for pkg in client.service.list_packages(args.project_dir):
+ print(pkg)
- @classmethod
- def register(cls, action):
- cls.repoactiondict[action.tag] = action
-
- @classmethod
- def print_actions(cls):
- print('available subcommands are:', file=sys.stderr)
- for a in cls.repoactiondict:
- print(f' {a}', file=sys.stderr)
-
- def __new__(cls, node):
- action = cls.repoactiondict[node]
- return object.__new__(action)
-
- def execute(self, _client, _opt, _args):
- raise NotImplementedError('execute() not implemented')
-
-
-class ListPackagesAction(RepoAction):
-
- tag = 'list_packages'
-
- def execute(self, client, _opt, args):
- if len(args) != 1:
- print(
- 'usage: elbe prjrepo list_packages <project_dir>',
- file=sys.stderr)
- sys.exit(199)
-
- builddir = args[0]
- for pkg in client.service.list_packages(builddir):
- print(pkg)
+ at add_argument('project_dir')
+def _download(client, args):
+ filename = 'repo.tar.gz'
+ client.service.tar_prjrepo(args.project_dir, filename)
-RepoAction.register(ListPackagesAction)
+ dst_fname = os.path.join(
+ '.',
+ 'elbe-projectrepo-' +
+ datetime.now().strftime('%Y%m%d-%H%M%S') +
+ '.tar.gz')
+ client.download_file(args.project_dir, filename, dst_fname)
+ print(f'{dst_fname} saved')
-class DownloadAction(RepoAction):
- tag = 'download'
+def _upload_file(client, f, builddir):
+ # Uploads file f into builddir in intivm
+ size = 1024 * 1024
+ part = 0
- def execute(self, client, _opt, args):
- if len(args) != 1:
- print('usage: elbe prjrepo download <project_dir>',
- file=sys.stderr)
- sys.exit(200)
-
- builddir = args[0]
- filename = 'repo.tar.gz'
- client.service.tar_prjrepo(builddir, filename)
-
- dst_fname = os.path.join(
- '.',
- 'elbe-projectrepo-' +
- datetime.now().strftime('%Y%m%d-%H%M%S') +
- '.tar.gz')
-
- client.download_file(builddir, filename, dst_fname)
- print(f'{dst_fname} saved')
-
-
-RepoAction.register(DownloadAction)
-
-
-class UploadPackageAction(RepoAction):
-
- tag = 'upload_pkg'
+ with open(f, 'rb') as fp:
+ while True:
- @staticmethod
- def upload_file(client, f, builddir):
- # Uploads file f into builddir in intivm
- size = 1024 * 1024
- part = 0
- with open(f, 'rb') as fp:
- while True:
-
- xml_base64 = binascii.b2a_base64(fp.read(size))
-
- if not isinstance(xml_base64, str):
- xml_base64 = xml_base64.decode('ascii')
-
- # finish upload
- if len(xml_base64) == 1:
- part = client.service.upload_file(builddir,
- os.path.basename(f),
- xml_base64,
- -1)
- else:
- part = client.service.upload_file(builddir,
- os.path.basename(f),
- xml_base64,
- part)
- if part == -1:
- print('project busy, upload not allowed')
- return -1
- if part == -2:
- print('Upload of package finished.')
- break
-
- def execute(self, client, _opt, args):
- if len(args) != 2:
- print(
- 'usage: elbe prjrepo upload_pkg <project_dir> <deb/dsc/changes file>',
- file=sys.stderr)
- sys.exit(201)
-
- builddir = args[0]
- filename = args[1]
-
- print('\n--------------------------')
- print('Upload and Include Package')
- print('--------------------------')
- print('Check files...')
-
- filetype = os.path.splitext(filename)[1]
-
- # Check filetype
- if filetype not in ['.dsc', '.deb', '.changes']:
- print('Error: Only .dsc, .deb or .changes files allowed to upload.')
- sys.exit(202)
-
- files = [filename] # list of all files which will be uploaded
-
- # Parse .dsc-File and append neccessary source files to files
- if filetype == '.dsc':
- for f in debian.deb822.Dsc(open(filename))['Files']:
- files.append(f['name'])
-
- if filetype == '.changes':
- for f in debian.deb822.Changes(open(filename))['Files']:
- files.append(f['name'])
-
- # Check whether all files are available
- abort = False
- for f in files:
- if not os.path.isfile(f):
- print(f'File {f} not found.')
- abort = True
- # Abort if one or more source files are missing
- if abort:
- sys.exit(203)
-
- print('Start uploading file(s)...')
- for f in files:
- print(f'Upload {f}...')
- self.upload_file(client, f, builddir)
-
- print('Including Package in initvm...')
- client.service.include_package(builddir, os.path.basename(filename))
-
-
-RepoAction.register(UploadPackageAction)
+ xml_base64 = binascii.b2a_base64(fp.read(size))
+
+ if not isinstance(xml_base64, str):
+ xml_base64 = xml_base64.decode('ascii')
+
+ # finish upload
+ if len(xml_base64) == 1:
+ part = client.service.upload_file(builddir,
+ os.path.basename(f),
+ xml_base64,
+ -1)
+ else:
+ part = client.service.upload_file(builddir,
+ os.path.basename(f),
+ xml_base64,
+ part)
+ if part == -1:
+ print('project busy, upload not allowed')
+ return -1
+ if part == -2:
+ print('Upload of package finished.')
+ break
+
+
+ at add_argument('project_dir')
+ at add_argument('package')
+def _upload_pkg(client, args):
+ print('\n--------------------------')
+ print('Upload and Include Package')
+ print('--------------------------')
+ print('Check files...')
+
+ filetype = os.path.splitext(args.package)[1]
+
+ # Check filetype
+ if filetype not in ['.dsc', '.deb', '.changes']:
+ print('Error: Only .dsc, .deb or .changes files allowed to upload.')
+ sys.exit(202)
+
+ files = [args.package] # list of all files which will be uploaded
+
+ # Parse .dsc-File and append neccessary source files to files
+ if filetype == '.dsc':
+ for f in debian.deb822.Dsc(open(args.package))['Files']:
+ files.append(f['name'])
+
+ if filetype == '.changes':
+ for f in debian.deb822.Changes(open(args.package))['Files']:
+ files.append(f['name'])
+
+ # Check whether all files are available
+ abort = False
+ for f in files:
+ if not os.path.isfile(f):
+ print(f'File {f} not found.')
+ abort = True
+ # Abort if one or more source files are missing
+ if abort:
+ sys.exit(203)
+
+ print('Start uploading file(s)...')
+ for f in files:
+ print(f'Upload {f}...')
+ _upload_file(client, f, args.project_dir)
+
+ print('Including Package in initvm...')
+ client.service.include_package(args.project_dir, os.path.basename(args.package))
+
+
+repo_actions = {
+ 'list_packages': _list_packages,
+ 'download': _download,
+ 'upload_pkg': _upload_pkg,
+}
--
2.45.2
More information about the elbe-devel
mailing list