forked from pyvec/python.cz
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathgithub.py
More file actions
125 lines (95 loc) · 3.32 KB
/
Copy pathgithub.py
File metadata and controls
125 lines (95 loc) · 3.32 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
# -*- coding: utf-8 -*-
import itertools
from datetime import datetime
import requests
import grequests
from werkzeug.contrib.cache import FileSystemCache
from .. import app
__all__ = ('get_issues',)
cache = FileSystemCache(app.config['CACHE_DIR'], default_timeout=3600)
def get_issues(org_names):
issues = cache.get('github-issues')
if issues is None:
session = _create_github_api_session()
issues = itertools.chain(*(_get_issues_for_org(session, org)
for org in org_names))
# Using grequests here to get the issues details asynchronously
# Note that it does not support session, so we set the headers manually
reqs = (grequests.get(_issue_url(issue), headers=session.headers)
for issue in issues)
issues = [_enhance_issue(response) for response in grequests.map(reqs)]
issues = sorted(issues, key=_get_issue_sort_key, reverse=True)
cache.set('github-issues', issues)
return issues
def _create_github_api_session():
user_agent = ('pythoncz/{now.year}-{now.month} '
'(+https://python.cz)').format(now=datetime.now())
session = requests.Session()
session.headers.update({
'User-Agent': user_agent,
'Authorization': 'token {}'.format(app.config['GITHUB_TOKEN']),
'Accept': 'application/vnd.github.squirrel-girl-preview',
})
return session
def _get_issues_for_org(session, org_name):
search_conditions = [
'is:open',
'org:{}'.format(org_name)
]
page = 1
while True:
res = session.get('https://api.github.com/search/issues', params={
'q': ' '.join(search_conditions),
'per_page': 100,
'page': page,
})
res.raise_for_status()
items = res.json().get('items', [])
if items:
yield from items
page += 1
else:
break
def _issue_repo_slug(issue):
repo_url_segments = issue['repository_url'].split('/')
org, repo = repo_url_segments[-2], repo_url_segments[-1]
return org, repo, '{}/{}'.format(org, repo)
def _issue_url(issue):
org, repo, slug = _issue_repo_slug(issue)
return 'https://api.github.com/repos/{}/issues/{}'.format(
slug,
issue['number']
)
def _enhance_issue(res):
res.raise_for_status()
issue = res.json()
try:
is_pull_request = bool(issue['pull_request']['url'])
except KeyError:
is_pull_request = False
try:
reactions = issue['reactions']
issue['votes'] = (
reactions.get('total_count', 0)
- reactions.get('-1', 0)
- reactions.get('confused', 0)
)
except KeyError:
issue['votes'] = 0
labels = [label['name'] for label in issue.get('labels', [])]
org, repo, slug = _issue_repo_slug(issue)
issue['coach'] = 'coach' in labels
issue['is_pull_request'] = is_pull_request
issue['repository_name'] = repo
issue['organization_name'] = org
issue['repository_full_name'] = slug
issue['repository_url_html'] = 'https://github.com/' + slug
return issue
def _get_issue_sort_key(issue):
return (
not issue.get('assignee'),
issue.get('coach'),
issue.get('votes'),
issue.get('comments'),
issue.get('updated_at'),
)