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
|
from datetime import datetime
from pydantic import BaseModel, Field, parse_obj_as
from pydis_core.site_api import APIClient
class NominationEntry(BaseModel):
"""Pydantic model representing a nomination entry."""
actor_id: int = Field(alias="actor")
reason: str
inserted_at: datetime
class Nomination(BaseModel):
"""Pydantic model representing a nomination."""
id: int
active: bool
user_id: int = Field(alias="user")
inserted_at: datetime
end_reason: str
ended_at: datetime | None
entries: list[NominationEntry]
reviewed: bool
thread_id: int | None
class NominationAPI:
"""Abstraction of site API interaction for talentpool."""
def __init__(self, site_api: APIClient):
self.site_api = site_api
async def get_nominations(
self,
user_id: int | None = None,
active: bool | None = None,
ordering: str = "-inserted_at"
) -> list[Nomination]:
"""
Fetch a list of nominations.
Passing a value of `None` indicates it shouldn't filtered by.
"""
params = {"ordering": ordering}
if active is not None:
params["active"] = str(active)
if user_id is not None:
params["user__id"] = str(user_id)
data = await self.site_api.get("bot/nominations", params=params)
nominations = parse_obj_as(list[Nomination], data)
return nominations
async def get_nomination(self, nomination_id: int) -> Nomination:
"""Fetch a nomination by ID."""
data = await self.site_api.get(f"bot/nominations/{nomination_id}")
nomination = Nomination.parse_obj(data)
return nomination
async def edit_nomination(
self,
nomination_id: int,
*,
end_reason: str | None = None,
active: bool | None = None,
reviewed: bool | None = None,
thread_id: int | None = None,
) -> Nomination:
"""
Edit a nomination.
Passing a value of `None` indicates it shouldn't be updated.
"""
data = {}
if end_reason is not None:
data["end_reason"] = end_reason
if active is not None:
data["active"] = active
if reviewed is not None:
data["reviewed"] = reviewed
if thread_id is not None:
data["thread_id"] = thread_id
result = await self.site_api.patch(f"bot/nominations/{nomination_id}", json=data)
return Nomination.parse_obj(result)
async def edit_nomination_entry(
self,
nomination_id: int,
*,
actor_id: int,
reason: str,
) -> Nomination:
"""Edit a nomination entry."""
data = {"actor": actor_id, "reason": reason}
result = await self.site_api.patch(f"bot/nominations/{nomination_id}", json=data)
return Nomination.parse_obj(result)
async def post_nomination(
self,
user_id: int,
actor_id: int,
reason: str,
) -> Nomination:
"""Post a nomination to site."""
data = {
"actor": actor_id,
"reason": reason,
"user": user_id,
}
result = await self.site_api.post("bot/nominations", json=data)
return Nomination.parse_obj(result)
async def get_activity(
self,
user_ids: list[int],
*,
days: int,
) -> dict[int, int]:
"""
Get the number of messages sent in the past `days` days by users with the given IDs.
Returns a dictionary mapping user ID to message count.
"""
result = await self.site_api.post(
"bot/users/metricity_activity_data",
json=user_ids,
params={"days": str(days)}
)
return {int(user_id): message_count for user_id, message_count in result.items()}
|
