typing: Make things stricter (4/4)

Add annotations to all calls.

Signed-off-by: Stephen Finucane <stephen@that.guru>

Author: stephenfin

git_pw/api.py

@@ -22,7 +22,7 @@
 
 FC = TypeVar('FC', bound=Callable[..., Any])
 
-Filters = list[tuple[str, str]]
+Filters = list[tuple[str, str | int | None]]
 
 
 class HTTPTokenAuth(requests.auth.AuthBase):
@@ -236,7 +236,7 @@ def version() -> tuple[int, int]:
 
 def get(url: str, params: Filters | None = None) -> dict[str, Any]:
     """Get a JSON document from the API and return it as a dict."""
-    return _get(url, params, stream=False).json()
+    return cast(dict[str, Any], _get(url, params, stream=False).json())
 
 
 def download(
@@ -323,7 +323,7 @@ def index(
     params = params or []
     params.append(('project', _get_project()))
 
-    return _get(url, params).json()
+    return cast(list[dict[str, Any]], _get(url, params).json())
 
 
 def detail(
@@ -346,7 +346,7 @@ def detail(
     # NOTE(stephenfin): All resources must have a trailing '/'
     url = '/'.join([_get_server(), resource_type, str(resource_id), ''])
 
-    return _get(url, params, stream=False).json()
+    return cast(dict[str, Any], _get(url, params, stream=False).json())
 
 
 def create(
@@ -367,7 +367,7 @@ def create(
     # NOTE(stephenfin): All resources must have a trailing '/'
     url = '/'.join([_get_server(), resource_type, ''])
 
-    return _post(url, data).json()
+    return cast(dict[str, Any], _post(url, data).json())
 
 
 def delete(resource_type: str, resource_id: str | int) -> None:
@@ -408,38 +408,38 @@ def update(
     # NOTE(stephenfin): All resources must have a trailing '/'
     url = '/'.join([_get_server(), resource_type, str(resource_id), ''])
 
-    return _patch(url, data).json()
+    return cast(dict[str, Any], _patch(url, data).json())
 
 
 def validate_minimum_version(
     min_version: tuple[int, int],
     msg: str,
-) -> Callable[[Any], Any]:
-    def inner(f):
+) -> Callable[[FC], FC]:
+    def inner(f: FC) -> FC:
         @click.pass_context
-        def new_func(ctx, *args, **kwargs):
+        def new_func(ctx: click.Context, *args: Any, **kwargs: Any) -> Any:
             if version() < min_version:
                 LOG.error(msg)
                 sys.exit(1)
 
             return ctx.invoke(f, *args, **kwargs)
 
-        return update_wrapper(new_func, f)
+        return cast(FC, update_wrapper(new_func, f))
 
     return inner
 
 
 def validate_multiple_filter_support(f: FC) -> FC:
     @click.pass_context
-    def new_func(ctx, *args, **kwargs):
+    def new_func(ctx: click.Context, *args: Any, **kwargs: Any) -> Any:
         if version() >= (1, 1):
             return ctx.invoke(f, *args, **kwargs)
 
         for param in ctx.command.params:
             if not param.multiple:
                 continue
 
-            if param.name in ('headers'):
+            if param.name is None or param.name == 'headers':
                 continue
 
             value = list(kwargs[param.name] or [])

git_pw/bundle.py

@@ -86,9 +86,9 @@ def download_cmd(bundle_id: str, output: str | None) -> None:
         LOG.info('Downloaded bundle to %s', path)
 
 
-def _show_bundle(bundle: dict[str, Any], fmt: str) -> None:
-    def _format_patch(patch):
-        return '%-4d %s' % (patch.get('id'), patch.get('name'))
+def _show_bundle(bundle: dict[str, Any], fmt: str | None) -> None:
+    def _format_patch(patch: dict[str, Any]) -> str:
+        return '%-4d %s' % (patch['id'], patch.get('name'))
 
     output = [
         ('ID', bundle.get('id')),
@@ -137,7 +137,15 @@ def show_cmd(fmt: str, bundle_id: str) -> None:
 @utils.format_options(headers=_list_headers)
 @click.argument('name', required=False)
 @api.validate_multiple_filter_support
-def list_cmd(owners, limit, page, sort, fmt, headers, name):
+def list_cmd(
+    owners: tuple[str, ...],
+    limit: int | None,
+    page: int | None,
+    sort: str,
+    fmt: str | None,
+    headers: tuple[str, ...],
+    name: str | None,
+) -> None:
     """List bundles.
 
     List bundles on the Patchwork instance.
@@ -150,7 +158,7 @@ def list_cmd(owners, limit, page, sort, fmt, headers, name):
         sort,
     )
 
-    params = []
+    params: list[tuple[str, str | int | None]] = []
 
     for owner in owners:
         # we support server-side filtering by username (but not email) in 1.1

git_pw/patch.py

@@ -2,6 +2,7 @@
 Patch subcommands.
 """
 
+import datetime
 import logging
 import os
 import sys
@@ -70,7 +71,12 @@ def _get_apply_patch_deps() -> bool:
     ),
 )
 @click.argument('args', nargs=-1, type=click.UNPROCESSED)
-def apply_cmd(patch_id, series, deps, args):
+def apply_cmd(
+    patch_id: int,
+    series: int | None,
+    deps: bool,
+    args: tuple[str, ...],
+) -> None:
     """Apply patch.
 
     Apply a patch locally using the 'git-am' command. Any additional ARGS
@@ -86,14 +92,17 @@ def apply_cmd(patch_id, series, deps, args):
 
     patch = api.detail('patches', patch_id)
 
+    series_filter: int | str | None = series
     if deps and not series:
-        series = '*'
+        series_filter = '*'
     elif not deps:
-        series = None
+        series_filter = None
 
     mbox = api.download(
         patch['mbox'],
-        [('series', str(series))] if series is not None else None,
+        [('series', str(series_filter))]
+        if series_filter is not None
+        else None,
     )
 
     if mbox:
@@ -117,7 +126,7 @@ def apply_cmd(patch_id, series, deps, args):
     default=True,
     help='Show patch in mbox format.',
 )
-def download_cmd(patch_id, output, fmt):
+def download_cmd(patch_id: int, output: str | None, fmt: str) -> None:
     """Download patch in diff or mbox format.
 
     Download a patch but do not apply it. ``OUTPUT`` is optional and can be an
@@ -131,6 +140,7 @@ def download_cmd(patch_id, output, fmt):
 
     if fmt == 'diff':
         if output and not os.path.isdir(output):
+            output_path: int | str
             if output == '-':
                 output_path = 0  # stdout fd
             else:
@@ -154,9 +164,9 @@ def download_cmd(patch_id, output, fmt):
         LOG.info('Downloaded patch to %s', path)
 
 
-def _show_patch(patch, fmt):
-    def _format_series(series):
-        return '%-4d %s' % (series.get('id'), series.get('name') or '-')
+def _show_patch(patch: dict[str, Any], fmt: str | None) -> None:
+    def _format_series(series: dict[str, Any]) -> str:
+        return '%-4d %s' % (series['id'], series.get('name') or '-')
 
     output = [
         ('ID', patch.get('id')),
@@ -167,17 +177,17 @@ def _format_series(series):
         (
             'Submitter',
             '{} ({})'.format(
-                patch.get('submitter').get('name'),
-                patch.get('submitter').get('email'),
+                (patch.get('submitter') or {}).get('name'),
+                (patch.get('submitter') or {}).get('email'),
             ),
         ),
         ('State', patch.get('state')),
         ('Archived', patch.get('archived')),
-        ('Project', patch.get('project').get('name')),
+        ('Project', (patch.get('project') or {}).get('name')),
         (
             'Delegate',
             (
-                patch.get('delegate').get('username')
+                (patch.get('delegate') or {}).get('username')
                 if patch.get('delegate')
                 else ''
             ),
@@ -186,7 +196,7 @@ def _format_series(series):
     ]
 
     prefix = 'Series'
-    for series in patch.get('series'):
+    for series in patch.get('series') or []:
         output.append((prefix, _format_series(series)))
         prefix = ''
 
@@ -196,7 +206,7 @@ def _format_series(series):
 @click.command(name='show')
 @utils.format_options
 @click.argument('patch_id', type=click.INT)
-def show_cmd(fmt, patch_id):
+def show_cmd(fmt: str | None, patch_id: int) -> None:
     """Show information about patch.
 
     Retrieve Patchwork metadata for a patch.
@@ -208,7 +218,7 @@ def show_cmd(fmt, patch_id):
     _show_patch(patch, fmt)
 
 
-def _get_states():
+def _get_states() -> list[str] | tuple[str, ...]:
     return CONF.states.split(',') if CONF.states else _default_states
 
 
@@ -244,7 +254,14 @@ def _get_states():
     help='Set the patch archived state.',
 )
 @utils.format_options
-def update_cmd(patch_ids, commit_ref, state, delegate, archived, fmt):
+def update_cmd(
+    patch_ids: tuple[int, ...],
+    commit_ref: str | None,
+    state: str | None,
+    delegate: str | None,
+    archived: bool | None,
+    fmt: str | None,
+) -> None:
     """Update one or more patches.
 
     Updates one or more Patches on the Patchwork instance. Some operations may
@@ -339,20 +356,20 @@ def update_cmd(patch_ids, commit_ref, state, delegate, archived, fmt):
 @click.argument('name', required=False)
 @api.validate_multiple_filter_support
 def list_cmd(
-    states,
-    submitters,
-    delegates,
-    hashes,
-    archived,
-    since,
-    before,
-    limit,
-    page,
-    sort,
-    fmt,
-    headers,
-    name,
-):
+    states: tuple[str, ...],
+    submitters: tuple[str, ...],
+    delegates: tuple[str, ...],
+    hashes: tuple[str, ...],
+    archived: bool,
+    since: datetime.datetime | None,
+    before: datetime.datetime | None,
+    limit: int | None,
+    page: int | None,
+    sort: str,
+    fmt: str | None,
+    headers: tuple[str, ...],
+    name: str | None,
+) -> None:
     """List patches.
 
     List patches on the Patchwork instance.
@@ -367,7 +384,7 @@ def list_cmd(
         archived,
     )
 
-    params = []
+    params: list[tuple[str, str | int | None]] = []
 
     for state in states:
         params.append(('state', state))

git_pw/series.py

@@ -2,6 +2,7 @@
 Series subcommands.
 """
 
+import datetime
 import logging
 import os.path
 import sys
@@ -34,7 +35,7 @@
     'first.',
 )
 @click.argument('args', nargs=-1, type=click.UNPROCESSED)
-def apply_cmd(series_id, args, deps):
+def apply_cmd(series_id: int, args: tuple[str, ...], deps: bool) -> None:
     """Apply series.
 
     Apply a series locally using the 'git-am' command. Any additional ARGS
@@ -91,7 +92,7 @@ def apply_cmd(series_id, args, deps):
     default=True,
     help='Download all series patches to one file',
 )
-def download_cmd(series_id, output, fmt):
+def download_cmd(series_id: int, output: str | None, fmt: str) -> None:
     """Download series in mbox format.
 
     Download a series but do not apply it. ``OUTPUT`` is optional and can be an
@@ -131,7 +132,7 @@ def download_cmd(series_id, output, fmt):
 @click.command(name='show')
 @utils.format_options
 @click.argument('series_id', type=click.INT)
-def show_cmd(fmt, series_id):
+def show_cmd(fmt: str | None, series_id: int) -> None:
     """Show information about series.
 
     Retrieve Patchwork metadata for a series.
@@ -140,8 +141,8 @@ def show_cmd(fmt, series_id):
 
     series = api.detail('series', series_id)
 
-    def _format_submission(submission):
-        return '%-4d %s' % (submission.get('id'), submission.get('name'))
+    def _format_submission(submission: dict[str, Any]) -> str:
+        return '%-4d %s' % (submission['id'], submission.get('name'))
 
     output = [
         ('ID', series.get('id')),
@@ -165,7 +166,7 @@ def _format_submission(submission):
         (
             'Cover',
             (
-                _format_submission(series.get('cover_letter'))
+                _format_submission(series['cover_letter'])
                 if series.get('cover_letter')
                 else ''
             ),
@@ -195,7 +196,17 @@ def _format_submission(submission):
 @utils.format_options(headers=_list_headers)
 @click.argument('name', required=False)
 @api.validate_multiple_filter_support
-def list_cmd(submitters, limit, page, sort, fmt, headers, name, since, before):
+def list_cmd(
+    submitters: tuple[str, ...],
+    limit: int | None,
+    page: int | None,
+    sort: str,
+    fmt: str | None,
+    headers: tuple[str, ...],
+    name: str | None,
+    since: datetime.datetime | None,
+    before: datetime.datetime | None,
+) -> None:
     """List series.
 
     List series on the Patchwork instance.
@@ -208,7 +219,7 @@ def list_cmd(submitters, limit, page, sort, fmt, headers, name, since, before):
         sort,
     )
 
-    params = []
+    params: list[tuple[str, str | int | None]] = []
 
     for submitter in submitters:
         if submitter.isdigit():