reonokiy/swh-lister
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add anotation type in some lister file

Browse source
This commit is contained in:
Gautier Pugnonblanc Yann 2020-02-17 15:58:34 +01:00
parent 73a33d9224
commit 60adc424be
12 changed files with 148 additions and 85 deletions
Download patch file Download diff file Expand all files Collapse all files

23
swh/lister/github/lister.py
View file

@ -5,11 +5,13 @@
import re
from typing import Any
from typing import Any, Dict, List, Tuple, Optional
from swh.lister.core.indexing_lister import IndexingHttpLister
from swh.lister.github.models import GitHubModel
from requests import Response
class GitHubLister(IndexingHttpLister):
PATH_TEMPLATE = '/repositories?since=%d'
@ -20,7 +22,7 @@ class GitHubLister(IndexingHttpLister):
instance = 'github' # There is only 1 instance of such lister
default_min_bound = 0 # type: Any
def get_model_from_repo(self, repo):
def get_model_from_repo(self, repo: Dict[str, Any]) -> Dict[str, Any]:
return {
'uid': repo['id'],
'indexable': repo['id'],
@ -32,7 +34,7 @@ class GitHubLister(IndexingHttpLister):
'fork': repo['fork'],
}
def transport_quota_check(self, response):
def transport_quota_check(self, response: Response) -> Tuple[bool, int]:
x_rate_limit_remaining = response.headers.get('X-RateLimit-Remaining')
if not x_rate_limit_remaining:
return False, 0
@ -42,17 +44,21 @@ class GitHubLister(IndexingHttpLister):
return True, delay
return False, 0
def get_next_target_from_response(self, response):
def get_next_target_from_response(self,
response: Response) -> Optional[int]:
if 'next' in response.links:
next_url = response.links['next']['url']
return int(self.API_URL_INDEX_RE.match(next_url).group(1))
return int(
self.API_URL_INDEX_RE.match(next_url).group(1)) # type: ignore
return None
def transport_response_simplified(self, response):
def transport_response_simplified(self, response: Response
) -> List[Dict[str, Any]]:
repos = response.json()
return [self.get_model_from_repo(repo)
for repo in repos if repo and 'id' in repo]
def request_headers(self):
def request_headers(self) -> Dict[str, Any]:
"""(Override) Set requests headers to send when querying the GitHub API
"""
@ -60,7 +66,8 @@ class GitHubLister(IndexingHttpLister):
headers['Accept'] = 'application/vnd.github.v3+json'
return headers
def disable_deleted_repo_tasks(self, index, next_index, keep_these):
def disable_deleted_repo_tasks(self, index: int,
next_index: int, keep_these: int):
""" (Overrides) Fix provided index value to avoid erroneously disabling
some scheduler tasks
"""