Add an --app= flag for specifying the WSGI application

docs/runner.rst

@@ -20,6 +20,10 @@ Is equivalent to::
 
     waitress-serve --port=8041 --url-scheme=https myapp:wsgifunc
 
+Or:
+
+    waitress-serve --port=8041 --url-scheme=https --app=myapp:wsgifunc
+
 The full argument list is :ref:`given below <invocation>`.
 
 Boolean arguments are represented by flags. If you wish to explicitly set a
@@ -64,13 +68,19 @@ Invocation
 
 Usage::
 
-    waitress-serve [OPTS] MODULE:OBJECT
+    waitress-serve [OPTS] [MODULE:OBJECT]
 
 Common options:
 
 ``--help``
     Show this information.
 
+``--app=MODULE:OBJECT``
+    Run the given callable object the WSGI application.
+
+    You can specify the WSGI application using this flag or as a positional
+    argument.
+
 ``--call``
     Call the given object to get the WSGI application.
 

src/waitress/adjustments.py

@@ -14,6 +14,7 @@
 """Adjustments are tunable parameters.
 """
 import getopt
+import pkgutil
 import socket
 import warnings
 
@@ -95,9 +96,25 @@ class _int_marker(int):
     pass
 
 
+class AppResolutionError(Exception):
+    """The named WSGI application could not be resolved."""
+
+
+def resolve_wsgi_app(app_name, call=False):
+    """Resolve a WSGI app descriptor to a callable."""
+    try:
+        app = pkgutil.resolve_name(app_name)
+    except (ValueError, ImportError, AttributeError) as exc:
+        raise AppResolutionError(f"Cannot import WSGI application '{app_name}': {exc}")
+    return app() if call else app
+
+
 class Adjustments:
     """This class contains tunable parameters."""
 
+    # If you add new parameters, be sure to update the following files:
+    #  * src/arguments.rst (waitress.serve)
+    #  * src/waitress/runner.py and src/runner.rst (CLI documentation)
     _params = (
         ("host", str),
         ("port", int),
@@ -459,11 +476,15 @@ def parse_args(cls, argv):
             else:
                 long_opts.append(opt + "=")
 
+        long_opts.append("app=")
+
         kw = {
             "help": False,
             "call": False,
+            "app": None,
         }
 
+        app = None
         opts, args = getopt.getopt(argv, "", long_opts)
         for opt, value in opts:
             param = opt.lstrip("-").replace("-", "_")
@@ -477,12 +498,25 @@ def parse_args(cls, argv):
                 kw[param] = "false"
             elif param in ("help", "call"):
                 kw[param] = True
+            elif param == "app":
+                app = value
             elif cls._param_map[param] is asbool:
                 kw[param] = "true"
             else:
                 kw[param] = value
 
-        return kw, args
+        if not kw["help"]:
+            if app is None and len(args) > 0:
+                app = args.pop(0)
+            if app is None:
+                raise AppResolutionError("Specify an application")
+            if len(args) > 0:
+                raise AppResolutionError("Provide only one WSGI app")
+            kw["app"] = resolve_wsgi_app(app, kw["call"])
+
+        del kw["call"]
+
+        return kw
 
     @classmethod
     def check_sockets(cls, sockets):

src/waitress/runner.py

@@ -11,31 +11,35 @@
 # FOR A PARTICULAR PURPOSE.
 #
 ##############################################################################
-"""Command line runner.
-"""
-
+"""Command line runner."""
 
 import getopt
 import logging
 import os
 import os.path
-import pkgutil
 import sys
+import traceback
 
 from waitress import serve
-from waitress.adjustments import Adjustments
+from waitress.adjustments import Adjustments, AppResolutionError
 from waitress.utilities import logger
 
 HELP = """\
 Usage:
 
-    {0} [OPTS] MODULE:OBJECT
+    {0} [OPTS] [MODULE:OBJECT]
 
 Standard options:
 
     --help
         Show this information.
 
+    --app=MODULE:OBJECT
+        Run the given callable object the WSGI application.
+
+        You can specify the WSGI application using this flag or as a positional
+        argument.
+
     --call
         Call the given object to get the WSGI application.
 
@@ -277,62 +281,37 @@ def show_help(stream, name, error=None):  # pragma: no cover
     print(HELP.format(name), file=stream)
 
 
-def show_exception(stream):
-    exc_type, exc_value = sys.exc_info()[:2]
-    args = getattr(exc_value, "args", None)
-    print(
-        ("There was an exception ({}) importing your module.\n").format(
-            exc_type.__name__,
-        ),
-        file=stream,
-    )
-    if args:
-        print("It had these arguments: ", file=stream)
-        for idx, arg in enumerate(args, start=1):
-            print(f"{idx}. {arg}\n", file=stream)
-    else:
-        print("It had no arguments.", file=stream)
-
-
 def run(argv=sys.argv, _serve=serve):
     """Command line runner."""
+    # Add the current directory onto sys.path
+    sys.path.append(os.getcwd())
+
     name = os.path.basename(argv[0])
 
     try:
-        kw, args = Adjustments.parse_args(argv[1:])
+        kw = Adjustments.parse_args(argv[1:])
     except getopt.GetoptError as exc:
         show_help(sys.stderr, name, str(exc))
         return 1
+    except AppResolutionError as exc:
+        show_help(sys.stderr, name, str(exc))
+        traceback.print_exc(file=sys.stderr)
+        return 1
 
     if kw["help"]:
         show_help(sys.stdout, name)
         return 0
 
-    if len(args) != 1:
-        show_help(sys.stderr, name, "Specify one application only")
-        return 1
-
     # set a default level for the logger only if it hasn't been set explicitly
     # note that this level does not override any parent logger levels,
     # handlers, etc but without it no log messages are emitted by default
     if logger.level == logging.NOTSET:
         logger.setLevel(logging.INFO)
 
-    # Add the current directory onto sys.path
-    sys.path.append(os.getcwd())
-
-    # Get the WSGI function.
-    try:
-        app = pkgutil.resolve_name(args[0])
-    except (ValueError, ImportError, AttributeError) as exc:
-        show_help(sys.stderr, name, str(exc))
-        show_exception(sys.stderr)
-        return 1
-    if kw["call"]:
-        app = app()
+    app = kw["app"]
 
     # These arguments are specific to the runner, not waitress itself.
-    del kw["call"], kw["help"]
+    del kw["help"], kw["app"]
 
     _serve(app, **kw)
     return 0

tests/test_adjustments.py

@@ -395,38 +395,57 @@ def assertDictContainsSubset(self, subset, dictionary):
         self.assertTrue(set(subset.items()) <= set(dictionary.items()))
 
     def test_noargs(self):
-        opts, args = self.parse([])
-        self.assertDictEqual(opts, {"call": False, "help": False})
-        self.assertSequenceEqual(args, [])
+        from waitress.adjustments import AppResolutionError
+
+        self.assertRaises(AppResolutionError, self.parse, [])
 
     def test_help(self):
-        opts, args = self.parse(["--help"])
-        self.assertDictEqual(opts, {"call": False, "help": True})
-        self.assertSequenceEqual(args, [])
+        opts = self.parse(["--help"])
+        self.assertDictEqual(opts, {"help": True, "app": None})
+
+    def test_app_flag(self):
+        from tests.fixtureapps import runner as _apps
+
+        opts = self.parse(["--app=tests.fixtureapps.runner:app"])
+        self.assertEqual(opts["app"], _apps.app)
 
     def test_call(self):
-        opts, args = self.parse(["--call"])
-        self.assertDictEqual(opts, {"call": True, "help": False})
-        self.assertSequenceEqual(args, [])
+        from tests.fixtureapps import runner as _apps
+
+        opts = self.parse(["--app=tests.fixtureapps.runner:returns_app", "--call"])
+        self.assertEqual(opts["app"], _apps.app)
 
-    def test_both(self):
-        opts, args = self.parse(["--call", "--help"])
-        self.assertDictEqual(opts, {"call": True, "help": True})
-        self.assertSequenceEqual(args, [])
+    def test_app_arg(self):
+        from tests.fixtureapps import runner as _apps
+
+        opts = self.parse(["tests.fixtureapps.runner:app"])
+        self.assertEqual(opts["app"], _apps.app)
+
+    def test_excess(self):
+        from waitress.adjustments import AppResolutionError
+
+        self.assertRaises(
+            AppResolutionError,
+            self.parse,
+            ["tests.fixtureapps.runner:app", "tests.fixtureapps.runner:app"],
+        )
 
     def test_positive_boolean(self):
-        opts, args = self.parse(["--expose-tracebacks"])
+        opts = self.parse(["--expose-tracebacks", "tests.fixtureapps.runner:app"])
         self.assertDictContainsSubset({"expose_tracebacks": "true"}, opts)
-        self.assertSequenceEqual(args, [])
 
     def test_negative_boolean(self):
-        opts, args = self.parse(["--no-expose-tracebacks"])
+        opts = self.parse(["--no-expose-tracebacks", "tests.fixtureapps.runner:app"])
         self.assertDictContainsSubset({"expose_tracebacks": "false"}, opts)
-        self.assertSequenceEqual(args, [])
 
     def test_cast_params(self):
-        opts, args = self.parse(
-            ["--host=localhost", "--port=80", "--unix-socket-perms=777"]
+        opts = self.parse(
+            [
+                "--host=localhost",
+                "--port=80",
+                "--unix-socket-perms=777",
+                "tests.fixtureapps.runner:app",
+            ]
         )
         self.assertDictContainsSubset(
             {
@@ -436,28 +455,25 @@ def test_cast_params(self):
             },
             opts,
         )
-        self.assertSequenceEqual(args, [])
 
     def test_listen_params(self):
-        opts, args = self.parse(
+        opts = self.parse(
             [
                 "--listen=test:80",
+                "tests.fixtureapps.runner:app",
             ]
         )
-
         self.assertDictContainsSubset({"listen": " test:80"}, opts)
-        self.assertSequenceEqual(args, [])
 
     def test_multiple_listen_params(self):
-        opts, args = self.parse(
+        opts = self.parse(
             [
                 "--listen=test:80",
                 "--listen=test:8080",
+                "tests.fixtureapps.runner:app",
             ]
         )
-
         self.assertDictContainsSubset({"listen": " test:80 test:8080"}, opts)
-        self.assertSequenceEqual(args, [])
 
     def test_bad_param(self):
         import getopt