Perform type checking with mypy

And fix some weird typing mistakes.
parent 14309f07
Pipeline #54896003 failed with stages
in 20 minutes and 8 seconds
......@@ -7,3 +7,4 @@ coverage*
.coverage
.eggs
build
.mypy_cache
......@@ -6,9 +6,10 @@ lint:
tags: [docker]
script:
- apk add --no-cache --update python3-dev gcc build-base
- pip install flake8 pylint
- pip install flake8 pylint mypy
- flake8 hydra_cli tests
- pylint hydra_cli
- mypy hydra_cli
test:
stage: test
......
......@@ -8,7 +8,7 @@ buildPythonApplication {
src = if lib.inNixShell then null else ./.;
propagatedBuildInputs = [ requests tabulate ];
checkInputs = [ nose nose-cov flake8 pylint autopep8 ];
checkInputs = [ nose nose-cov flake8 pylint autopep8 mypy ];
shellHook = ''
root_dir=$(git rev-parse --show-toplevel)
......
......@@ -2,7 +2,7 @@
Utilities to communicate with the Hydra API.
"""
from typing import List, Dict
from typing import List, Dict, Any
import re
from requests import Response, get
from hydra_cli.tables import get_desired_fields
......@@ -35,7 +35,7 @@ def perform_api_call_json(
property_path: str,
fields: List[str],
is_single: bool,
) -> Dict[str, List[any]]:
) -> Dict[str, List[Any]]:
"""
Performs an API call to a given Hydra server, ensures a JSON response with
a configurable set of fields is returned and handles possible errors.
......@@ -49,7 +49,7 @@ def perform_api_call_json(
def perform_api_call(
host: str,
property_path: str,
headers: List[str] = {}
headers: Dict[str, str] = {}
) -> Response:
"""
Performs a raw API call to a Hydra, ensures that the request actually comes from a Hydra
......
......@@ -3,7 +3,8 @@ Simple helper which deals provides utility to locally
cache files in ~/.cache.
"""
import os
from typing import Optional, Callable, Iterable
from typing import Optional, Callable
from requests import Response
class Cache():
......@@ -23,7 +24,7 @@ class Cache():
if not os.path.exists(self.cache_dir):
os.mkdir(self.cache_dir)
def ensure_cache_entry(self, cache_path: str, provider: Callable[[], Iterable[str]], mode: int):
def ensure_cache_entry(self, cache_path: str, provider: Callable[[], Response], mode: int):
"""
Ensures that a given cache path exists and optionally invokes the `provider`
which returns an iterable object (i.e. a response from `requests`) and its content
......
......@@ -2,7 +2,7 @@
CLI Application parses `argv` and decides what to do.
"""
import argparse
from typing import List
from typing import List, Any, Dict
import sys
import os
from urllib.parse import urlparse
......@@ -103,31 +103,34 @@ class Application():
else:
raise err
def project(self, argv: List[str]):
def project(self, argv: Dict[str, Any]):
"""
Show available build projects from Hydra (mainly list projects or show details).
"""
fields = argv.fields
if argv.verbose:
fields = argv['fields']
if argv['verbose']:
fields = self.project_choices
is_single = argv.show is not None
path = "/project/%s" % argv.show if is_single else "/"
is_single = argv['show'] is not None
path = "/project/%s" % argv['show'] if is_single else "/"
print(render_table(argv.plain, perform_api_call_json(argv.host, path, fields, is_single)))
print(render_table(
argv['plain'],
perform_api_call_json(argv['host'], path, fields, is_single)
))
def reproduce_build(self, argv: List[str]):
def reproduce_build(self, argv: Dict[str, Any]):
"""
Calls the reproduce build functionality from Hydra for a given build ID.
"""
host = '{uri.netloc}'.format(uri=urlparse(argv.host))
script_path = 'reproduce-build-%s-%d' % (host, argv.build_nr)
host = '{uri.netloc}'.format(uri=urlparse(argv['host']))
script_path = 'reproduce-build-%s-%d' % (host, argv['build_nr'])
self.cache.ensure_cache_entry(
script_path,
lambda: perform_api_call(argv.host, '/build/%d/reproduce' % argv.build_nr),
lambda: perform_api_call(argv['host'], '/build/%d/reproduce' % argv['build_nr']),
0o775
)
......
"""
Collection of general utility for table rendering and transformation.
"""
from typing import List, Dict
from typing import List, Dict, Any, Tuple
from tabulate import tabulate
def get_desired_fields(json: Dict[str, any], fields: List[str]) -> Dict[str, List[any]]:
def get_desired_fields(json: List[Dict[str, Any]], fields: List[str]) -> Dict[str, List[Any]]:
"""
Fetches desired fields from a list of objects describing a Hydra
object (e.g. a jobset or a project).
......@@ -23,7 +23,7 @@ def get_desired_fields(json: Dict[str, any], fields: List[str]) -> Dict[str, Lis
return {f: list(map(lambda x: x[f], json)) for f in fields}
def render_table(plain: bool, table: Dict[str, List[any]]) -> str:
def render_table(plain: bool, table: Dict[str, List[Any]]) -> str:
"""
Renders the dict by creating a table using `tabulate` and
transforming object names to readable entities using `.title()`.
......@@ -39,7 +39,7 @@ def render_table(plain: bool, table: Dict[str, List[any]]) -> str:
)
def flatten_arrays(key: str, data: List[any]) -> List[any]:
def flatten_arrays(key: str, data: List[Any]) -> Tuple[str, List[Any]]:
"""
Flattens arrays in a row, so they're rendered comma-separated rather
than in the classical array form `['foo', 'bar', 'baz']`.
......
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