Browse Source

Use argparse config throughout app (#349)

This PR is a pretty substantial refactor of the entrypoints of pypiserver (`__main__` and `__init__`) to use the argparse-based config added in #339.

- Updated `RunConfig` and `UpdateConfig` classes to have exclusive init kwargs, instead of taking an namespace. This turned out to be much easier when working with the library-style app initialization in `__init__`, both for direct instantiation and via paste config
- Added an `iter_packages()` method to the `RunConfig` to iterate over packages specified by the configuration (note @elfjes, I think that replacing this with e.g. a `backend` reference will be a nice way to tie in #348)
- Added a general-purpose method to map legacy keyword arguments to the `app()` and `paste_app_factory()` functions to updated forms
- Refactored the `paste_app_factory()` to not mutate the incoming dictionary
- Removed all argument-parsing and config-related code from `__main__` and `core`
- Moved `_logwrite` from `__init__` to `__main__`, since that was the only place it was being used after the updates to `core`
- Updated `digest_file` to use `` instead of `getattr(hashlib, algo)`, because the former supports more algorithms
- Updated `` to, instead of calling `eval()` on the entirety of `__init__`, to instead just evaluate the line that defines the version
- Assigned the config to a `._pypiserver_config` attribute on the `Bottle` instance to reduce hacky test workarounds
- Fixed the tox config, which I broke in #339 

* Config: add auth & absolute path resolution

* Config: check pkg dirs on config creation

* Instantiate config with kwargs, not namespace

* WIP: still pulling the threads

* Init seems to be working

* tests passing locally, still need to update cache

* Fix tox command

* unused import

* Fix typing

* Be more selective in exec() in

* Require accurate casing for hash algos

* Remove old comment

* Comments, minor updates and simplifications

* move _logwrite to a more reasonable place

* Update config to work with cache

* Type cachemanager listdir in core

* Update config module docstring, rename method

* Add more comments re: paste config

* Add comments to main, remove unneded check

* Remove commented code

* Use {posargs} instead of [] for clarity in tox

* Add dupe check for kwarg updater

* Remove unused references on app instance

* Fix typo

* Remove redundancy in log level parsing
Matthew Planchard 2 years ago committed by GitHub
No known key found for this signature in database
  1. 363
  2. 353
  3. 31
  4. 364
  5. 118
  6. 11
  7. 5
  8. 10
  9. 28
  10. 107
  11. 19
  12. 99
  13. 110
  14. 18
  15. 15


@ -1,6 +1,11 @@
import os
import functools
import pathlib
import re as _re
import sys
import typing as t
from pypiserver.bottle import Bottle
from pypiserver.config import Config, RunConfig, strtobool
version = __version__ = "2.0.0dev1"
__version_info__ = tuple(_re.split("[.-]", __version__))
@ -11,203 +16,191 @@ __summary__ = "A minimal PyPI server for use with pip/easy_install."
__uri__ = ""
class Configuration:
.. see:: config-options: :func:`pypiserver.configure()`
identity = lambda x: x
def __init__(self, **kwds):
def __repr__(self, *args, **kwargs):
return f"Configuration(**{vars(self)})"
def backwards_compat_kwargs(kwargs: dict, warn: bool = True) -> dict:
"""Return a dict with deprecated kwargs converted to new kwargs.
def __str__(self, *args, **kwargs):
return "Configuration:\n" + "\n".join(
f"{k:>20} = {v}" for k, v in sorted(vars(self).items())
:param kwargs: the incoming kwargs to convert
:param warn: whether to output a warning to stderr if there are deprecated
kwargs found in the incoming kwargs
# A mapping of deprecated kwargs to a 2-tuple of their corresponding updated
# kwarg and a function to convert the value of the deprecated kwarg to a
# value for the new kwarg. `identity` is just a function that returns
# whatever it is passed and is used in cases where the only change from
# a legacy kwarg is its name.
backwards_compat = {
"authenticated": ("authenticate", identity),
"passwords": ("password_file", identity),
# `root` could be a string or an array of strings. Handle both cases,
# converting strings to Path instances.
"root": (
lambda root: [
# Convert strings to absolute Path instances
for r in ([root] if isinstance(root, str) else root)
# `redirect_to_fallback` was changed to `disable_fallback` for clearer
# use as a flag to disable the default behavior. Since its behavior
# is the opposite, we negate it.
"redirect_to_fallback": (
lambda redirect: not redirect,
"server": ("server_method", identity),
# `welcome_msg` now is just provided as text, so that anyone using
# pypiserver as a library doesn't need to worry about distributing
# files if they don't need to. If we're still passed an old-style
# `welcome_file` argument, we go ahead and resolve it to an absolute
# path and read the text.
"welcome_file": (
lambda p: pathlib.Path(p).expanduser().resolve().read_text(),
# Warn the user if they're using any deprecated arguments
if warn and any(k in backwards_compat for k in kwargs):
# Make nice instructions like `Please replace the following:
# 'authenticated' with 'authenticate'` and print to stderr.
replacement_strs = (
f"'{k}' with '{backwards_compat[k][0]}'"
for k in filter(lambda k: k in kwargs, backwards_compat)
warn_str = (
"You are using deprecated arguments. Please replace the following: \n"
f" {', '.join(replacement_strs)}"
print(warn_str, file=sys.stderr)
# Create an iterable of 2-tuple to collect into the updated dictionary. Each
# item will either be the existing key-value pair from kwargs, or, if the
# keyword is a legacy keyword, the new key and potentially adjusted value
# for that keyword. Note that depending on the order the argument are
# specified, this _could_ mean an updated legacy keyword could override
# a new argument if that argument is also specified. However, in that
# case, our updated kwargs dictionary would have a different number of
# keys compared to our incoming dictionary, so we check for that case
# below.
rv_iter = (
(k, v)
if k not in backwards_compat
else (backwards_compat[k][0], backwards_compat[k][1](v))
for k, v in kwargs.items()
updated_kwargs = dict(rv_iter)
# If our dictionaries have different lengths, we must have gotten duplicate
# legacy/modern keys. Figure out which keys were dupes and throw an error.
if len(updated_kwargs) != len(kwargs):
legacy_to_modern = {k: v[0] for k, v in backwards_compat.items()}
dupes = [
(k, v)
for k, v in legacy_to_modern.items()
if k in kwargs and v in kwargs
raise ValueError(
"Keyword arguments for pypiserver app() constructor contained "
"duplicate legacy and modern keys. Duplicates are shown below, in "
"the form (legacy_key, modern_key):\n"
def update(self, props):
d = props if isinstance(props, dict) else vars(props)
def default_config(
log_err_frmt="%(body)s: %(exception)s \n%(traceback)s",
Fetch default-opts with overridden kwds, capable of starting-up pypiserver.
Does not validate overridden options.
Example usage::
kwds = pypiserver.default_config(<override_kwds> ...)
## More modifications on kwds.**kwds)``.
Kwds correspond to same-named cmd-line opts, with '-' --> '_' substitution.
Non standard args are described below:
:param return_defaults_only:
When `True`, returns defaults, otherwise,
configures "runtime" attributes and returns also the "packages"
found in the roots.
:param root:
A list of paths, derived from the packages specified on cmd-line.
If `None`, defaults to '~/packages'.
:param redirect_to_fallback:
see :option:`--disable-fallback`
:param authenticated:
see :option:`--authenticate`
:param password_file:
see :option:`--passwords`
:param log_file:
see :option:`--log-file`
Not used, passed here for logging it.
:param log_frmt:
see :option:`--log-frmt`
Not used, passed here for logging it.
:param callable auther:
An API-only options that if it evaluates to a callable,
it is invoked to allow access to protected operations
(instead of htpaswd mechanism) like that::
auther(username, password): bool
When defined, `password_file` is ignored.
:param host:
see :option:`--interface`
Not used, passed here for logging it.
:param port:
see :option:`--port`
Not used, passed here for logging it.
:param server:
see :option:`--server`
Not used, passed here for logging it.
:param verbosity:
see :option:`-v`
Not used, passed here for logging it.
:param VERSION:
Not used, passed here for logging it.
:return: a dict of defaults
return updated_kwargs
return locals()
def app(**kwargs: t.Any) -> Bottle:
"""Construct a bottle app running pypiserver.
def app(**kwds):
:param kwds: Any overrides for defaults. Any property of RunConfig
(or its base), defined in `pypiserver.config`, may be overridden.
:param dict kwds: Any overrides for defaults, as fetched by
:func:`default_config()`. Check the docstring of this function
for supported kwds.
from . import core
config = Config.default_with_overrides(**backwards_compat_kwargs(kwargs))
return app_from_config(config)
def app_from_config(config: RunConfig) -> Bottle:
"""Construct a bottle app from the provided RunConfig."""
# The _app module instantiates a Bottle instance directly when it is
# imported. That is ``. We directly mutate some global variables
# on the imported `_app` module so that its endpoints will behave as
# we expect.
_app = __import__("_app", globals(), locals(), ["."], 1)
# Because we're about to mutate our import, we pop it out of the imported
# modules map, so that any future imports do not receive our mutated version
sys.modules.pop("pypiserver._app", None)
kwds = default_config(**kwds)
config, packages = core.configure(**kwds)
_app.config = config
_app.packages = packages = _app # HACK for testing.
# Add a reference to our config on the Bottle app for easy access in testing
# and other contexts. = config
def str2bool(s, default):
if s is not None and s != "":
return s.lower() not in ("no", "off", "0", "false")
return default
def _str_strip(string):
"""Provide a generic strip method to pass as a callback."""
return string.strip()
def paste_app_factory(global_config, **local_conf):
"""Parse a paste config and return an app."""
def upd_conf_with_bool_item(conf, attr, sdict):
conf[attr] = str2bool(sdict.pop(attr, None), conf[attr])
def upd_conf_with_str_item(conf, attr, sdict):
value = sdict.pop(attr, None)
if value is not None:
conf[attr] = value
def upd_conf_with_int_item(conf, attr, sdict):
value = sdict.pop(attr, None)
if value is not None:
conf[attr] = int(value)
def upd_conf_with_list_item(conf, attr, sdict, sep=" ", parse=_str_strip):
values = sdict.pop(attr, None)
if values:
conf[attr] = list(filter(None, map(parse, values.split(sep))))
def _make_root(root):
root = root.strip()
if root.startswith("~"):
return os.path.expanduser(root)
return root
c = default_config()
upd_conf_with_bool_item(c, "overwrite", local_conf)
upd_conf_with_bool_item(c, "redirect_to_fallback", local_conf)
upd_conf_with_list_item(c, "authenticated", local_conf, sep=" ")
upd_conf_with_list_item(c, "root", local_conf, sep="\n", parse=_make_root)
upd_conf_with_int_item(c, "verbosity", local_conf)
str_items = [
for str_item in str_items:
upd_conf_with_str_item(c, str_item, local_conf)
# cache_control is undocumented; don't know what type is expected:
# upd_conf_with_str_item(c, 'cache_control', local_conf)
return app(**c)
def _logwrite(logger, level, msg):
if msg:
line_endings = ["\r\n", "\n\r", "\n"]
for le in line_endings:
if msg.endswith(le):
msg = msg[: -len(le)]
if msg:
logger.log(level, msg)
T = t.TypeVar("T")
def paste_app_factory(_global_config, **local_conf):
"""Parse a paste config and return an app.
The paste config is entirely strings, so we need to parse those
strings into values usable for the config, if they're present.
def to_bool(val: t.Optional[str]) -> t.Optional[bool]:
"""Convert a string value, if provided, to a bool."""
return val if val is None else strtobool(val)
def to_int(val: t.Optional[str]) -> t.Optional[int]:
"""Convert a string value, if provided, to an int."""
return val if val is None else int(val)
def to_list(
val: t.Optional[str],
sep: str = " ",
transform: t.Callable[[str], T] = str.strip,
) -> t.Optional[t.List[T]]:
"""Convert a string value, if provided, to a list.
:param sep: the separator between items in the string representation
of the list
:param transform: an optional function to call on each string item of
the list
if val is None:
return val
return list(filter(None, map(transform, val.split(sep))))
def _make_root(root: str) -> pathlib.Path:
"""Convert a specified string root into an absolute Path instance."""
return pathlib.Path(root.strip()).expanduser().resolve()
# A map of config keys we expect in the paste config to the appropriate
# function to parse the string config value. This map includes both
# current and legacy keys.
maps = {
"cache_control": to_int,
"roots": functools.partial(to_list, sep="\n", transform=_make_root),
# root is a deprecated argument for roots
"root": functools.partial(to_list, sep="\n", transform=_make_root),
"disable_fallback": to_bool,
# redirect_to_fallback is a deprecated argument for disable_fallback
"redirect_to_fallback": to_bool,
"overwrite": to_bool,
"authenticate": functools.partial(to_list, sep=" "),
# authenticated is a deprecated argument for authenticate
"authenticated": functools.partial(to_list, sep=" "),
"verbosity": to_int,
# First, convert values from strings to whatever types we need, or leave
# them as strings if there's no mapping function available for them.
mapped_conf = {k: maps.get(k, identity)(v) for k, v in local_conf.items()}
# Convert any legacy key/value pairs into their modern form.
updated_conf = backwards_compat_kwargs(mapped_conf)
return app(**updated_conf)


@ -3,14 +3,12 @@
from __future__ import print_function
import getopt
import logging
import os
import re
import sys
import textwrap
import typing as t
import functools as ft
from pypiserver.config import Config, UpdateConfig
log = logging.getLogger("pypiserver.main")
@ -20,7 +18,7 @@ def init_logging(
stream: t.Optional[t.IO] = sys.stderr,
logger = logger or logging.getLogger()
@ -38,332 +36,77 @@ def init_logging(
def usage():
return textwrap.dedent(
start PyPI compatible package server serving packages from
command line, it uses the default ~/packages. pypiserver scans this
directory recursively for packages. It skips packages and
directories starting with a dot. Multiple package directories can be
pypi-server understands the following options:
-p, --port PORT
Listen on port PORT (default: 8080).
-i, --interface INTERFACE
Listen on interface INTERFACE (default:, any interface).
-a, --authenticate (update|download|list), ...
Comma-separated list of (case-insensitive) actions to authenticate.
Requires to have set the password (-P option).
To password-protect package downloads (in addition to uploads) while
leaving listings public, use:
-P foo/htpasswd.txt -a update,download
To allow unauthorized access, use:
-P . -a .
Note that when uploads are not protected, the `register` command
is not necessary, but `~/.pypirc` still need username and password fields,
even if bogus.
By default, only 'update' is password-protected.
-P, --passwords PASSWORD_FILE
Use apache htpasswd file PASSWORD_FILE to set usernames & passwords when
authenticating certain actions (see -a option).
To allow unauthorized access, use:
-P . -a .
Disable redirect to real PyPI index for packages not found in the
local index.
--fallback-url FALLBACK_URL
For packages not found in the local index, this URL will be used to
redirect to (default:
--server METHOD
Use METHOD to run the server. Valid values include paste,
cherrypy, twisted, gunicorn, gevent, wsgiref, auto. The
default is to use "auto" which chooses one of paste, cherrypy,
twisted or wsgiref.
[deprecated] Serve packages from PACKAGES_DIRECTORY.
-o, --overwrite
Allow overwriting existing package files.
--hash-algo ALGO
Any `hashlib` available algo used as fragments on package links.
Set one of (0, no, off, false) to disabled it (default: md5).
--welcome HTML_FILE
Uses the ASCII contents of HTML_FILE as welcome message response.
Enable verbose logging; repeat for more verbosity.
--log-file FILE
Write logging info into this FILE, as well as to stdout or stderr, if configured.
--log-stream STREAM
Log messages to the specified STREAM. Valid values are "stdout", "stderr", or "none"
--log-frmt FORMAT
The logging format-string. (see `logging.LogRecord` class from standard python library)
[Default: %(asctime)s|%(name)s|%(levelname)s|%(thread)d|%(message)s]
--log-req-frmt FORMAT
A format-string selecting Http-Request properties to log; set to '%s' to see them all.
[Default: %(bottle.request)s]
--log-res-frmt FORMAT
A format-string selecting Http-Response properties to log; set to '%s' to see them all.
[Default: %(status)s]
--log-err-frmt FORMAT
A format-string selecting Http-Error properties to log; set to '%s' to see them all.
[Default: %(body)s: %(exception)s \n%(traceback)s]
--cache-control AGE
Add "Cache-Control: max-age=AGE, public" header to package downloads.
Pip 6+ needs this for caching.
pypi-server -h, --help
Show this help message.
pypi-server --version
Show pypi-server's version.
Update packages in PACKAGES_DIRECTORY. This command searches for updates and shows a pip command line which
updates the package.
The following additional options can be specified with -U:
Execute the pip commands instead of only showing them.
Download package updates to this directory. The default is to use
the directory which contains the latest version of the package to
be updated.
Allow updating to unstable version (alpha, beta, rc, dev versions).
--blacklist-file BLACKLIST_FILE
Don't update packages listed in this file (one package name per line,
without versions, '#' comments honored). This can be useful if you upload
private packages into pypiserver, but also keep a mirror of public
packages that you regularly update. Attempting to pull an update of
a private package from `` might pose a security risk - e.g. a
malicious user might publish a higher version of the private package,
containing arbitrary code.
Visit for more information.
def main(argv=None):
"""Application entrypoint for pypiserver.
This function drives the application (as opposed to the library)
implementation of pypiserver. Usage from the commandline will result in
this function being called.
import pypiserver
if argv is None:
argv = sys.argv
# The first item in sys.argv is the name of the python file being
# executed, which we don't need
argv = sys.argv[1:]
command = "serve"
c = pypiserver.Configuration(**pypiserver.default_config())
update_dry_run = True
update_directory = None
update_stable_only = True
update_blacklist_file = None
opts, roots = getopt.getopt(
except getopt.GetoptError:
err = sys.exc_info()[1]
sys.exit(f"usage error: {err}")
for k, v in opts:
if k in ("-p", "--port"):
c.port = int(v)
except Exception:
err = sys.exc_info()[1]
sys.exit(f"Invalid port({v!r}) due to: {err}")
elif k in ("-a", "--authenticate"):
c.authenticated = [
a.lower() for a in re.split("[, ]+", v.strip(" ,")) if a
if c.authenticated == ["."]:
c.authenticated = []
actions = ("list", "download", "update")
for a in c.authenticated:
if a not in actions:
errmsg = (
f"Action '{a}' for option `{k}`"
f" not one of {actions}!"
elif k in ("-i", "--interface"): = v
elif k in ("-r", "--root"):
elif k == "--disable-fallback":
c.redirect_to_fallback = False
elif k == "--fallback-url":
c.fallback_url = v
elif k == "--server":
c.server = v
elif k == "--welcome":
c.welcome_file = v
elif k == "--version":
print(f"pypiserver {pypiserver.__version__}\n")
elif k == "-U":
command = "update"
elif k == "-x":
update_dry_run = False
elif k == "-u":
update_stable_only = False
elif k == "-d":
update_directory = v
elif k == "--blacklist-file":
update_blacklist_file = v
elif k in ("-P", "--passwords"):
c.password_file = v
elif k in ("-o", "--overwrite"):
c.overwrite = True
elif k == "--hash-algo":
c.hash_algo = None if not pypiserver.str2bool(v, c.hash_algo) else v
elif k == "--log-file":
c.log_file = v
elif k == "--log-stream":
c.log_stream = v
elif k == "--log-frmt":
c.log_frmt = v
elif k == "--log-req-frmt":
c.log_req_frmt = v
elif k == "--log-res-frmt":
c.log_res_frmt = v
elif k == "--log-err-frmt":
c.log_err_frmt = v
elif k == "--cache-control":
c.cache_control = v
elif k == "-v":
c.verbosity += 1
elif k in ("-h", "--help"):
if (
not c.authenticated
and c.password_file != "."
or c.authenticated
and c.password_file == "."
"When auth-ops-list is empty (-a=.), password-file"
f" (-P={c.password_file!r}) must also be empty ('.')!"
if len(roots) == 0:
roots = [os.path.abspath(x) for x in roots]
c.root = roots
verbose_levels = [
log_level = list(zip(verbose_levels, range(c.verbosity)))[-1][0]
valid_streams = {"none": None, "stderr": sys.stderr, "stdout": sys.stdout}
if c.log_stream not in valid_streams:
f"Invalid log stream {c.log_stream}."
f" Choose one of {', '.join(valid_streams.keys())}"
config = Config.from_args(argv)
if command == "update":
# Check to see if we were asked to run an update command instead of running
# the server
if isinstance(config, UpdateConfig):
from pypiserver.manage import update_all_packages
dry_run=not config.execute,
# Fixes #49:
# The gevent server adapter needs to patch some
# modules BEFORE importing bottle!
if c.server and c.server.startswith("gevent"):
if config.server_method.startswith("gevent"):
import gevent.monkey # @UnresolvedImport
from pypiserver import bottle
if c.server not in bottle.server_names:
f"Unknown server {c.server}."
f" Choose one of {', '.join(bottle.server_names.keys())}"
bottle.debug(c.verbosity > 1)
bottle.debug(config.verbosity > 1)
bottle._stderr = ft.partial(
pypiserver._logwrite, logging.getLogger(bottle.__name__), logging.INFO
_logwrite, logging.getLogger(bottle.__name__), logging.INFO
# Here `app` is a Bottle instance, which we pass to to run
# the server
app = pypiserver.app_from_config(config)
app =**vars(c)),, port=c.port, server=c.server)
def _logwrite(logger, level, msg):
if msg:
line_endings = ["\r\n", "\n\r", "\n"]
for le in line_endings:
if msg.endswith(le):
msg = msg[: -len(le)]
if msg:
logger.log(level, msg)
if __name__ == "__main__":


@ -6,6 +6,7 @@ import re
import zipfile
import xml.dom.minidom
from pypiserver.config import RunConfig
from . import __version__
from . import core
from .bottle import (
@ -35,8 +36,7 @@ except ImportError: # PY2
log = logging.getLogger(__name__)
packages = None
config = None
config: RunConfig
app = Bottle()
@ -49,7 +49,7 @@ class auth:
def __call__(self, method):
def protector(*args, **kwargs):
if self.action in config.authenticated:
if self.action in config.authenticate:
if not request.auth or request.auth[1] is None:
raise HTTPError(
401, headers={"WWW-Authenticate": 'Basic realm="pypi"'}
@ -104,8 +104,9 @@ def root():
fp = request.custom_fullpath
numpkgs = len(list(packages()))
numpkgs = len(list(config.iter_packages()))
except Exception as exc:
log.error(f"Could not list packages: {exc}")
numpkgs = 0
# Ensure template() does not consider `msg` as filename!
@ -150,7 +151,7 @@ def remove_pkg():
pkgs = list(
lambda pkg: pkg.pkgname == name and pkg.version == version,
if len(pkgs) == 0:
@ -186,7 +187,9 @@ def file_upload():
raise HTTPError(400, f"Bad filename: {uf.raw_filename}")
if not config.overwrite and core.exists(packages.root, uf.raw_filename):
if not config.overwrite and core.exists(
config.package_root, uf.raw_filename
f"Cannot upload {uf.raw_filename!r} since it already exists! \n"
" You may start server with `--overwrite` option. "
@ -197,7 +200,7 @@ def file_upload():
" You may start server with `--overwrite` option.",
), uf.raw_filename,, uf.raw_filename,
if request.auth:
user = request.auth[0]
@ -254,7 +257,7 @@ def handle_rpc():
response = []
ordering = 0
for p in packages():
for p in config.iter_packages():
if p.pkgname.count(value) > 0:
# We do not presently have any description/summary, returning
# version instead
@ -275,7 +278,7 @@ def handle_rpc():
def simpleindex():
links = sorted(core.get_prefixes(packages()))
links = sorted(core.get_prefixes(config.iter_packages()))
tmpl = """\
@ -301,11 +304,11 @@ def simple(prefix=""):
return redirect("/simple/{0}/".format(normalized), 301)
files = sorted(
core.find_packages(packages(), prefix=prefix),
core.find_packages(config.iter_packages(), prefix=prefix),
key=lambda x: (x.parsed_version, x.relfn),
if not files:
if config.redirect_to_fallback:
if not config.disable_fallback:
return redirect(f"{config.fallback_url.rstrip('/')}/{prefix}/")
return HTTPError(404, f"Not Found ({normalized} does not exist)\n\n")
@ -338,7 +341,7 @@ def simple(prefix=""):
def list_packages():
fp = request.custom_fullpath
files = sorted(
key=lambda x: (os.path.dirname(x.relfn), x.pkgname, x.parsed_version),
links = [
@ -364,7 +367,7 @@ def list_packages():
def server_static(filename):
entries = core.find_packages(packages())
entries = core.find_packages(config.iter_packages())
for x in entries:
f = x.relfn_unix
if f == filename:


@ -6,42 +6,55 @@ the current config logic, but has not yet been integrated.
To add a config option:
- If it should be available for all subcommands (run, update), add it to
the `global_args` parser.
the `add_common_args()` function
- If it should only be available for the `run` command, add it to the
`run_parser` in the `get_parser()` function.
- If it should only be available for the `update` command, add it to the
`update_parser` in the `get_parser() function`.
- Add it to the appropriate Config class, `_ConfigCommon` for global options,
`RunConfig` for `run` options, and `UpdateConfig` for `update` options.
- This requires adding it as an `__init__()` kwarg, setting it as an instance
attribute in `__init__()`, and ensuring it will be parsed from the argparse
namespace in the `kwargs_from_namespace()` method
- Ensure your config option is tested in `tests/`.
The `Config` class provides a `.from_args()` static method, which returns
either a `RunConfig` or an `UpdateConfig`, depending on which subcommand
is specified in the args.
The `Config` class is a factory class only. Config objects do not inherit from
it, but from `_ConfigCommon`. The `Config` provides the following constructors:
- `default_with_overrides(**overrides: Any)`: construct a `RunConfig` (since
run is the default pypiserver action) with default values, applying any
specified overrides
- `from_args(args: Optional[Sequence[str]])`: construct a config from the
provided arguments. Depending on arguments, the config will be either a
`RunConfig` or an `UpdateConfig`
Legacy commandline arguments did not require a subcommand. This form is
still supported, but deprecated. A warning is printing to stderr if
the legacy commandline format is used.
Command line arguments should be parsed as early as possible, using
custom functions like the `auth_*` functions below if needed. For example,
if an option were to take JSON as an argument, that JSON should be parsed
into a dict by the argument parser.
import argparse
import contextlib
import hashlib
import itertools
import io
import itertools
import logging
import pathlib
import pkg_resources
import re
import textwrap
import sys
import textwrap
import typing as t
from distutils.util import strtobool as strtoint
from pypiserver import __version__
# The `passlib` requirement is optional, so we need to verify its import here.
from passlib.apache import HtpasswdFile
except ImportError:
HtpasswdFile = None
from pypiserver import core
# The "strtobool" function in distutils does a nice job at parsing strings,
@ -64,7 +77,7 @@ class DEFAULTS:
LOG_REQ_FRMT = "%(bottle.request)s"
LOG_RES_FRMT = "%(status)s"
LOG_STREAM = sys.stdout
PACKAGE_DIRECTORIES = ["~/packages"]
PACKAGE_DIRECTORIES = [pathlib.Path("~/packages").expanduser().resolve()]
PORT = 8080
@ -86,15 +99,14 @@ def auth_arg(arg: str) -> t.List[str]:
"Invalid authentication options. `.` (no authentication) "
"must be specified alone."
return items
# The "." is just an indicator for no auth, so we return an empty auth list
# if it was present.
return [i for i in items if not i == "."]
def hash_algo_arg(arg: str) -> t.Optional[str]:
"""Parse a hash algorithm from the string."""
# The standard hashing algorithms are all made available via fully
# lowercase names, along with (sometimes) variously cased versions
# as well.
arg = arg.lower()
if arg in hashlib.algorithms_available:
return arg
@ -130,9 +142,35 @@ def ignorelist_file_arg(arg: t.Optional[str]) -> t.List[str]:
"""Parse the ignorelist and return the list of ignored files."""
if arg is None or arg == "pypiserver/no-ignores":
return []
with open(arg) as f:
stripped_lines = (ln.strip() for ln in f.readlines())
return [ln for ln in stripped_lines if ln and not ln.startswith("#")]
fpath = pathlib.Path(arg)
if not fpath.exists():
raise argparse.ArgumentTypeError(f"No such ignorelist-file '{arg}'")
lines = (ln.strip() for ln in fpath.read_text().splitlines())
return [ln for ln in lines if ln and not ln.startswith("#")]
except Exception as exc:
raise argparse.ArgumentTypeError(
f"Could not parse ignorelist-file '{arg}': {exc}"
) from exc
def package_directory_arg(arg: str) -> pathlib.Path:
"""Convert any package directory argument into its absolute path."""
pkg_dir = pathlib.Path(arg).expanduser().resolve()
# Attempt to grab the first item from the directory. The directory may
# be empty, in which case we'll get back None, but if the directory does
# not exist or we do not have permission to read it, we can catch th
# OSError and exit with a useful message.
next(pkg_dir.iterdir(), None)
except OSError as exc:
raise argparse.ArgumentTypeError(
"Error: while trying to access package directory "
f"({pkg_dir}): {exc}"
return pkg_dir
# We need to capture this at compile time, because we replace sys.stderr
@ -161,6 +199,9 @@ def log_stream_arg(arg: str) -> t.Optional[t.IO]:
def add_common_args(parser: argparse.ArgumentParser) -> None:
"""Add common arguments to a parser."""
# Don't update at top-level to avoid circular imports in __init__
from pypiserver import __version__
@ -237,6 +278,7 @@ def get_parser() -> argparse.ArgumentParser:
help="The directory from which to serve packages.",
@ -249,7 +291,10 @@ def get_parser() -> argparse.ArgumentParser:
help="Listen on interface INTERFACE (default:",
@ -324,7 +369,7 @@ def get_parser() -> argparse.ArgumentParser:
"Use METHOD to run th eserver. Valid values include paste, "
"Use METHOD to run the server. Valid values include paste, "
"cherrypy, twisted, gunicorn, gevent, wsgiref, and auto. The "
'default is to use "auto", which chooses one of paste, cherrypy, '
"twisted, or wsgiref."
@ -412,6 +457,7 @@ def get_parser() -> argparse.ArgumentParser:
help="The directory from which to serve packages.",
@ -457,15 +503,83 @@ def get_parser() -> argparse.ArgumentParser:
return parser
TConf = t.TypeVar("TConf", bound="_ConfigCommon")
class _ConfigCommon:
def __init__(self, namespace: argparse.Namespace) -> None:
def __init__(
roots: t.List[pathlib.Path],
verbosity: int,
log_frmt: str,
log_file: t.Optional[str],
log_stream: t.Optional[t.IO],
) -> None:
"""Construct a RuntimeConfig."""
# Global arguments
self.verbosity: int = namespace.verbose
self.log_file: t.Optional[str] = namespace.log_file
self.log_stream: t.Optional[t.IO] = namespace.log_stream
self.log_frmt: str = namespace.log_frmt
self.roots: t.List[str] = namespace.package_directory
self.verbosity = verbosity
self.log_file = log_file
self.log_stream = log_stream
self.log_frmt = log_frmt
self.roots = roots
# Derived properties are directly based on other properties and are not
# included in equality checks.
self._derived_properties: t.Tuple[str, ...] = (
# The first package directory is considered the root. This is used
# for uploads.
self.package_root = self.roots[0]
def from_namespace(
cls: t.Type[TConf], namespace: argparse.Namespace
) -> TConf:
"""Construct a config from an argparse namespace."""
return cls(**cls.kwargs_from_namespace(namespace))
def kwargs_from_namespace(
namespace: argparse.Namespace,
) -> t.Dict[str, t.Any]:
"""Convert a namespace into __init__ kwargs for this class."""
return dict(
def log_level(self) -> int:
"""Return an appropriate log-level for the config's verbosity."""
levels = {
0: logging.WARNING,
1: logging.INFO,
2: logging.DEBUG,
# Return a log-level from warning through not set (log all messages).
# If we've specified 3 or more levels of verbosity, just return not set.
return levels.get(self.verbosity, logging.NOTSET)
def iter_packages(self) -> t.Iterator[core.PkgFile]:
"""Iterate over packages in root directories."""
yield from (
core.listdir(str(r)) for r in self.roots
def with_updates(self: TConf, **kwargs: t.Any) -> TConf:
"""Create a new config with the specified updates.
The current config is used as a base. Any properties not specified in
keyword arguments will remain unchanged.
return self.__class__(**{**dict(self), **kwargs}) # type: ignore
def __repr__(self) -> str:
"""A string representation indicating the class and its properties."""
@ -482,52 +596,176 @@ class _ConfigCommon:
"""Configs are equal if their public values are equal."""
if not isinstance(other, self.__class__):
return False
return all(getattr(other, k) == v for k, v in self) # type: ignore
return all(
getattr(other, k) == v
for k, v in self
if not k in self._derived_properties
def __iter__(self) -> t.Iterable[t.Tuple[str, t.Any]]:
def __iter__(self) -> t.Iterator[t.Tuple[str, t.Any]]:
"""Iterate over config (k, v) pairs."""
yield from (
(k, v) for k, v in vars(self).items() if not k.startswith("_")
(k, v)
for k, v in vars(self).items()
if not k.startswith("_") and k not in self._derived_properties
class RunConfig(_ConfigCommon):
"""A config for the Run command."""
def __init__(self, namespace: argparse.Namespace) -> None:
def __init__(
port: int,
host: str,
authenticate: t.List[str],
password_file: t.Optional[str],
disable_fallback: bool,
fallback_url: str,
server_method: str,
overwrite: bool,
hash_algo: t.Optional[str],
welcome_msg: str,
cache_control: t.Optional[int],
log_req_frmt: str,
log_res_frmt: str,
log_err_frmt: str,
auther: t.Callable[[str, str], bool] = None,
**kwargs: t.Any,
) -> None:
"""Construct a RuntimeConfig."""
self.port: int = namespace.port
self.interface: str = namespace.interface
self.authenticate: t.List[str] = namespace.authenticate
self.password_file: t.Optional[str] = namespace.passwords
self.disable_fallback: bool = namespace.disable_fallback
self.fallback_url: str = namespace.fallback_url
self.server_method: str = namespace.server
self.overwrite: bool = namespace.overwrite
self.hash_algo: t.Optional[str] = namespace.hash_algo
self.welcome_msg: str = namespace.welcome
self.cache_control: t.Optional[int] = namespace.cache_control
self.log_req_frmt: str = namespace.log_req_frmt
self.log_res_frmt: str = namespace.log_res_frmt
self.log_err_frmt: str = namespace.log_err_frmt
self.port = port = host
self.authenticate = authenticate
self.password_file = password_file
self.disable_fallback = disable_fallback
self.fallback_url = fallback_url
self.server_method = server_method
self.overwrite = overwrite
self.hash_algo = hash_algo
self.welcome_msg = welcome_msg
self.cache_control = cache_control
self.log_req_frmt = log_req_frmt
self.log_res_frmt = log_res_frmt
self.log_err_frmt = log_err_frmt
# Derived properties
self._derived_properties = self._derived_properties + ("auther",)
self.auther = self.get_auther(auther)
def kwargs_from_namespace(
cls, namespace: argparse.Namespace
) -> t.Dict[str, t.Any]:
"""Convert a namespace into __init__ kwargs for this class."""
return {
**super(RunConfig, cls).kwargs_from_namespace(namespace),
"port": namespace.port,
"authenticate": namespace.authenticate,
"password_file": namespace.passwords,
"disable_fallback": namespace.disable_fallback,
"fallback_url": namespace.fallback_url,
"server_method": namespace.server,
"overwrite": namespace.overwrite,
"hash_algo": namespace.hash_algo,
"welcome_msg": namespace.welcome,
"cache_control": namespace.cache_control,
"log_req_frmt": namespace.log_req_frmt,
"log_res_frmt": namespace.log_res_frmt,