Skip to content

Commit 4047c71

Browse files
ChrisLoveringChrisLovering
authored
Merge pull request #3309 from python-discord/update-pep-cog-2
Simplify PEP cog to use PEP API
2 parents 39df8e2 + 3568c5d commit 4047c71

File tree

1 file changed

+59
-123
lines changed

1 file changed

+59
-123
lines changed

‎bot/exts/info/pep.py

Lines changed: 59 additions & 123 deletions
Original file line numberDiff line numberDiff line change
@@ -1,163 +1,99 @@
11
from datetime import UTC, datetime, timedelta
2-
from email.parser import HeaderParser
3-
from io import StringIO
2+
from typing import TypedDict
43

54
from discord import Colour, Embed
65
from discord.ext.commands import Cog, Context, command
7-
from pydis_core.utils.caching import AsyncCache
86

97
from bot.bot import Bot
10-
from bot.constants import Keys
118
from bot.log import get_logger
129

1310
log = get_logger(__name__)
1411

1512
ICON_URL = "https://www.python.org/static/opengraph-icon-200x200.png"
16-
BASE_PEP_URL = "https://peps.python.org/pep-"
17-
PEPS_LISTING_API_URL = "https://api.github.com/repos/python/peps/contents/peps?ref=main"
13+
PEP_API_URL = "https://peps.python.org/api/peps.json"
1814

19-
pep_cache = AsyncCache()
15+
class PEPInfo(TypedDict):
16+
"""
17+
Useful subset of the PEP API response.
2018
21-
GITHUB_API_HEADERS = {}
22-
if Keys.github:
23-
GITHUB_API_HEADERS["Authorization"] = f"token {Keys.github}"
19+
Full structure documented at https://peps.python.org/api/
20+
"""
21+
22+
number: int
23+
title: str
24+
url: str
25+
status: str
26+
python_version: str | None
27+
created: str
28+
type: str
2429

2530

2631
class PythonEnhancementProposals(Cog):
2732
"""Cog for displaying information about PEPs."""
2833

2934
def __init__(self, bot: Bot):
3035
self.bot = bot
31-
self.peps: dict[int, str] = {}
32-
# Ensure peps are refreshed the first time this is checked
33-
self.last_refreshed_peps: datetime = datetime.min.replace(tzinfo=UTC)
34-
35-
async def refresh_peps_urls(self) -> None:
36-
"""Refresh PEP URLs listing in every 3 hours."""
37-
# Wait until HTTP client is available
38-
await self.bot.wait_until_ready()
39-
log.trace("Started refreshing PEP URLs.")
36+
self.peps: dict[int, PEPInfo] = {}
37+
self.last_refreshed_peps: datetime | None = None
38+
39+
async def refresh_pep_data(self) -> None:
40+
"""Refresh PEP data."""
41+
# Putting this first should prevent any race conditions
4042
self.last_refreshed_peps = datetime.now(tz=UTC)
4143

42-
async with self.bot.http_session.get(
43-
PEPS_LISTING_API_URL,
44-
headers=GITHUB_API_HEADERS
45-
) as resp:
44+
log.trace("Started refreshing PEP data.")
45+
async with self.bot.http_session.get(PEP_API_URL) as resp:
4646
if resp.status != 200:
47-
log.warning(f"Fetching PEP URLs from GitHub API failed with code {resp.status}")
47+
log.warning(
48+
f"Fetching PEP data from PEP API failed with code {resp.status}"
49+
)
4850
return
49-
5051
listing = await resp.json()
5152

52-
log.trace("Got PEP URLs listing from GitHub API")
53-
54-
for file in listing:
55-
name = file["name"]
56-
if name.startswith("pep-") and name.endswith((".rst", ".txt")):
57-
pep_number = name.replace("pep-", "").split(".")[0]
58-
self.peps[int(pep_number)] = file["download_url"]
59-
60-
log.info("Successfully refreshed PEP URLs listing.")
61-
62-
@staticmethod
63-
def get_pep_zero_embed() -> Embed:
64-
"""Get information embed about PEP 0."""
65-
pep_embed = Embed(
66-
title="**PEP 0 - Index of Python Enhancement Proposals (PEPs)**",
67-
url="https://peps.python.org/"
68-
)
69-
pep_embed.set_thumbnail(url=ICON_URL)
70-
pep_embed.add_field(name="Status", value="Active")
71-
pep_embed.add_field(name="Created", value="13-Jul-2000")
72-
pep_embed.add_field(name="Type", value="Informational")
53+
for pep_num, pep_info in listing.items():
54+
self.peps[int(pep_num)] = pep_info
7355

74-
return pep_embed
75-
76-
async def validate_pep_number(self, pep_nr: int) -> Embed | None:
77-
"""Validate is PEP number valid. When it isn't, return error embed, otherwise None."""
78-
if (
79-
pep_nr not in self.peps
80-
and (self.last_refreshed_peps + timedelta(minutes=30)) <= datetime.now(tz=UTC)
81-
and len(str(pep_nr)) < 5
82-
):
83-
await self.refresh_peps_urls()
84-
85-
if pep_nr not in self.peps:
86-
log.trace(f"PEP {pep_nr} was not found")
87-
return Embed(
88-
title="PEP not found",
89-
description=f"PEP {pep_nr} does not exist.",
90-
colour=Colour.red()
91-
)
56+
log.info("Successfully refreshed PEP data.")
9257

93-
return None
94-
95-
def generate_pep_embed(self, pep_header: dict, pep_nr: int) -> Embed:
96-
"""Generate PEP embed based on PEP headers data."""
97-
# the parsed header can be wrapped to multiple lines, so we need to make sure that is removed
98-
# for an example of a pep with this issue, see pep 500
99-
title = " ".join(pep_header["Title"].split())
100-
# Assemble the embed
101-
pep_embed = Embed(
102-
title=f"**PEP {pep_nr} - {title}**",
103-
url=f"{BASE_PEP_URL}{pep_nr:04}",
58+
def generate_pep_embed(self, pep: PEPInfo) -> Embed:
59+
"""Generate PEP embed."""
60+
embed = Embed(
61+
title=f"**PEP {pep['number']} - {pep['title']}**",
62+
url=pep["url"],
10463
)
64+
embed.set_thumbnail(url=ICON_URL)
10565

106-
pep_embed.set_thumbnail(url=ICON_URL)
66+
fields_to_check = ("status", "python_version", "created", "type")
67+
for field_name in fields_to_check:
68+
if field_value := pep.get(field_name):
69+
field_name = field_name.replace("_", " ").title()
70+
embed.add_field(name=field_name, value=field_value)
10771

108-
# Add the interesting information
109-
fields_to_check = ("Status", "Python-Version", "Created", "Type")
110-
for field in fields_to_check:
111-
# Check for a PEP metadata field that is present but has an empty value
112-
# embed field values can't contain an empty string
113-
if pep_header.get(field, ""):
114-
pep_embed.add_field(name=field, value=pep_header[field])
115-
116-
return pep_embed
117-
118-
@pep_cache(arg_offset=1)
119-
async def get_pep_embed(self, pep_nr: int) -> tuple[Embed, bool]:
120-
"""Fetch, generate and return PEP embed. Second item of return tuple show does getting success."""
121-
response = await self.bot.http_session.get(self.peps[pep_nr])
122-
123-
if response.status == 200:
124-
log.trace(f"PEP {pep_nr} found")
125-
pep_content = await response.text()
126-
127-
# Taken from https://github.com/python/peps/blob/master/pep0/pep.py#L179
128-
pep_header = HeaderParser().parse(StringIO(pep_content))
129-
return self.generate_pep_embed(pep_header, pep_nr), True
130-
131-
log.trace(
132-
f"The user requested PEP {pep_nr}, but the response had an unexpected status code: {response.status}."
133-
)
134-
return Embed(
135-
title="Unexpected error",
136-
description="Unexpected HTTP error during PEP search. Please let us know.",
137-
colour=Colour.red()
138-
), False
72+
return embed
13973

14074
@command(name="pep", aliases=("get_pep", "p"))
14175
async def pep_command(self, ctx: Context, pep_number: int) -> None:
14276
"""Fetches information about a PEP and sends it to the channel."""
143-
# Trigger typing in chat to show users that bot is responding
144-
await ctx.typing()
77+
# Refresh the PEP data up to every hour, as e.g. the PEP status might have changed.
78+
if (
79+
self.last_refreshed_peps is None or (
80+
(self.last_refreshed_peps + timedelta(hours=1)) <= datetime.now(tz=UTC)
81+
and len(str(pep_number)) < 5
82+
)
83+
):
84+
await self.refresh_pep_data()
14585

146-
# Handle PEP 0 directly because it's not in .rst or .txt so it can't be accessed like other PEPs.
147-
if pep_number == 0:
148-
pep_embed = self.get_pep_zero_embed()
149-
success = True
150-
else:
151-
success = False
152-
if not (pep_embed := await self.validate_pep_number(pep_number)):
153-
pep_embed, success = await self.get_pep_embed(pep_number)
154-
155-
await ctx.send(embed=pep_embed)
156-
if success:
157-
log.trace(f"PEP {pep_number} getting and sending finished successfully. Increasing stat.")
158-
self.bot.stats.incr(f"pep_fetches.{pep_number}")
86+
if pep := self.peps.get(pep_number):
87+
embed = self.generate_pep_embed(pep)
15988
else:
160-
log.trace(f"Getting PEP {pep_number} failed. Error embed sent.")
89+
log.trace(f"PEP {pep_number} was not found")
90+
embed = Embed(
91+
title="PEP not found",
92+
description=f"PEP {pep_number} does not exist.",
93+
colour=Colour.red(),
94+
)
95+
96+
await ctx.send(embed=embed)
16197

16298

16399
async def setup(bot: Bot) -> None:

0 commit comments

Comments
 (0)