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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
import logging
from json.decoder import JSONDecodeError
import httpx
from django.core.handlers.wsgi import WSGIRequest
from django.http import HttpResponse
from django.shortcuts import render
from django.utils import timezone
from django.views import View
from pydis_site import settings
from pydis_site.apps.home.models import RepositoryMetadata
log = logging.getLogger(__name__)
class HomeView(View):
"""The main landing page for the website."""
github_api = "https://api.github.com/users/python-discord/repos?per_page=100"
repository_cache_ttl = 3600
# Which of our GitHub repos should be displayed on the front page, and in which order?
repos = [
"python-discord/site",
"python-discord/bot",
"python-discord/snekbox",
"python-discord/sir-lancebot",
"python-discord/metricity",
"python-discord/king-arthur",
]
def __init__(self):
"""Clean up stale RepositoryMetadata."""
if not settings.STATIC_BUILD:
RepositoryMetadata.objects.exclude(repo_name__in=self.repos).delete()
# If no token is defined (for example in local development), then
# it does not make sense to pass the Authorization header. More
# specifically, GitHub will reject any requests from us due to the
# invalid header. We can make a limited number of anonymous requests
# though, which is useful for testing.
if settings.GITHUB_TOKEN:
self.headers = {"Authorization": f"token {settings.GITHUB_TOKEN}"}
else:
self.headers = {}
def _get_api_data(self) -> dict[str, dict[str, str]]:
"""
Call the GitHub API and get information about our repos.
If we're unable to get that info for any reason, return an empty dict.
"""
repo_dict = {}
try:
# Fetch the data from the GitHub API
resp = httpx.get(
self.github_api,
headers=self.headers,
timeout=settings.TIMEOUT_PERIOD
)
resp.raise_for_status()
api_data: list[dict] = resp.json()
except httpx.TimeoutException:
log.error("Request to fetch GitHub repository metadata for timed out!")
return repo_dict
except httpx.HTTPStatusError as ex:
log.error(f"Received HTTP {ex.response.status_code} from GitHub repository metadata request!")
return repo_dict
except JSONDecodeError:
log.error("GitHub returned invalid JSON for repository metadata!")
return repo_dict
# Process the API data into our dict
for repo in api_data:
try:
full_name = repo["full_name"]
if full_name in self.repos:
repo_dict[full_name] = {
"full_name": repo["full_name"],
"description": repo["description"],
"language": repo["language"],
"forks_count": repo["forks_count"],
"stargazers_count": repo["stargazers_count"],
}
# Something is not right about the API data we got back from GitHub.
except (TypeError, ConnectionError, KeyError) as e:
log.error(
"Unable to parse the GitHub repository metadata from response!",
extra={
'api_data': api_data,
'error': e
}
)
continue
return repo_dict
def _get_repo_data(self) -> list[RepositoryMetadata]:
"""Build a list of RepositoryMetadata objects that we can use to populate the front page."""
# First off, load the timestamp of the least recently updated entry.
if settings.STATIC_BUILD:
last_update = None
else:
last_update = (
RepositoryMetadata.objects.values_list("last_updated", flat=True)
.order_by("last_updated").first()
)
# If we did not retrieve any results here, we should import them!
if last_update is None:
# Try to get new data from the API. If it fails, we'll return an empty list.
# In this case, we simply don't display our projects on the site.
api_repositories = self._get_api_data()
# Create all the repodata records in the database.
data = [
RepositoryMetadata(
repo_name=api_data["full_name"],
description=api_data["description"],
forks=api_data["forks_count"],
stargazers=api_data["stargazers_count"],
language=api_data["language"],
)
for api_data in api_repositories.values()
]
if settings.STATIC_BUILD:
return data
return RepositoryMetadata.objects.bulk_create(data)
# If the data is stale, we should refresh it.
if (timezone.now() - last_update).seconds > self.repository_cache_ttl:
# Try to get new data from the API. If it fails, return the cached data.
api_repositories = self._get_api_data()
if not api_repositories:
return RepositoryMetadata.objects.all()
# Update or create all RepoData objects in self.repos
database_repositories = []
for api_data in api_repositories.values():
repo_data, _created = RepositoryMetadata.objects.update_or_create(
repo_name=api_data["full_name"],
defaults={
'repo_name': api_data["full_name"],
'description': api_data["description"],
'forks': api_data["forks_count"],
'stargazers': api_data["stargazers_count"],
'language': api_data["language"],
}
)
database_repositories.append(repo_data)
return database_repositories
# Otherwise, if the data is fresher than 2 minutes old, we should just return it.
return RepositoryMetadata.objects.all()
def get(self, request: WSGIRequest) -> HttpResponse:
"""Collect repo data and render the homepage view."""
repo_data = self._get_repo_data()
return render(request, "home/index.html", {"repo_data": repo_data})
|
