Python backend for a Slack's kudos plugin.
feat: events refactor
+42
-38
+5
kefi/constants.py
+5
kefi/constants.py
+5
-5
kefi/dependencies.py
+5
-5
kefi/dependencies.py
···
5
5
from kefi.models.database import engine
6
6
7
7
8
-
class InteractionParams:
9
-
def __init__(self, payload: str = Form(...)):
10
-
self.payload = payload
11
-
12
-
13
8
class Event(BaseModel):
14
9
type: str
15
10
user: str
···
20
15
challenge: str | None
21
16
type: str
22
17
event: Event | None
18
+
19
+
20
+
class InteractionParams:
21
+
def __init__(self, payload: str = Form(...)):
22
+
self.payload = payload
23
23
24
24
25
25
class SlashCommandParams:
+12
-1
kefi/models/core/helpers.py
+12
-1
kefi/models/core/helpers.py
···
2
2
from sqlmodel import Session, func, or_, select
3
3
4
4
from kefi.config import settings
5
-
from kefi.dependencies import SlashCommandParams
5
+
from kefi.dependencies import Event, SlashCommandParams
6
6
from kefi.models.database import Transaction, User
7
7
from kefi.routers.responses import ResetResponse
8
8
from kefi.slack import Slack
9
+
10
+
11
+
def get_or_create_from_event(event: Event, session: Session) -> tuple["User", bool]:
12
+
"""Gets or create the user from the given command."""
13
+
query = select(User).filter(User.slack_user_id == event.user)
14
+
user = session.exec(query).one_or_none()
15
+
created = not user
16
+
if not user:
17
+
user = User(slack_user_id=event.user)
18
+
session.add(user)
19
+
return (user, created)
9
20
10
21
11
22
def get_or_create_from_command(
+4
-8
kefi/routers/events.py
+4
-8
kefi/routers/events.py
···
1
1
from fastapi import APIRouter, Depends
2
2
from sqlmodel import Session
3
3
4
-
from kefi.dependencies import EventBody, InteractionParams, get_session
5
-
from kefi.routers.helpers import EventHandler, InteractionHandler
6
-
from kefi.routers.views import HomeView
7
-
from kefi.slack import Slack
4
+
from kefi.dependencies import EventBody, get_session
5
+
from kefi.routers.helpers import EventHandler
8
6
9
7
router = APIRouter()
10
8
11
9
12
10
@router.post("/events/", tags=["events"])
13
-
def handle_events(
14
-
body: EventBody,
15
-
):
11
+
def handle_events(event_body: EventBody, session: Session = Depends(get_session)):
16
12
"""Calls the events handler and gets the response."""
17
-
event_handler = EventHandler(body)
13
+
event_handler = EventHandler(event_body=event_body, session=session)
18
14
return event_handler.response()
+16
-20
kefi/routers/helpers.py
+16
-20
kefi/routers/helpers.py
···
4
4
5
5
from sqlmodel import Session
6
6
7
-
from kefi.constants import Command, InteractionType
8
-
from kefi.dependencies import (
9
-
EventBody,
10
-
EventType,
11
-
InteractionParams,
12
-
SlashCommandParams,
13
-
)
7
+
from kefi.constants import Command, EventBodyType, InteractionType
8
+
from kefi.dependencies import EventBody, InteractionParams, SlashCommandParams
14
9
from kefi.models.core.exceptions import NotEnoughKefi
15
10
from kefi.models.core.helpers import (
16
11
available_balance,
17
12
find_user_by_slack_user_id,
18
13
get_all_users,
19
14
get_or_create_from_command,
15
+
get_or_create_from_event,
20
16
get_or_create_from_interactivity,
21
17
received_balance,
22
18
send_admin_amount,
23
19
)
24
-
from kefi.models.database import User
25
20
from kefi.models.kudos.helpers import (
26
21
get_action,
27
22
notify_receiver_user_chat_action,
···
232
227
def interaction_view_submission(self) -> list | dict:
233
228
"""Handles the submissions from a view."""
234
229
view_id = self.payload["view"]["id"]
235
-
user_id = self.payload["user"]["id"]
236
230
view_callback_id = self.payload["view"]["callback_id"]
237
231
try:
238
232
return getattr(self, f"_view_submission_{view_callback_id}")(
···
249
243
class EventHandler:
250
244
def __init__(self, event_body: EventBody, session: Session):
251
245
self.event_body = event_body
252
-
246
+
self.session = session
253
247
self.slack = Slack()
254
-
self.user, _ = get_or_create_from_interactivity(
255
-
interactivity_payload=self.payload, session=session
256
-
)
248
+
if self.event_body.event:
249
+
self.user, _ = get_or_create_from_event(
250
+
event=self.event_body.event, session=session
251
+
)
257
252
258
-
def response(self) -> SlackResponse:
253
+
def response(self) -> list | dict:
259
254
handlers: dict[str, Callable] = {
260
-
"url_verification": self.handle_url_verification,
261
-
"event_callback": self.handle_event_callback,
255
+
EventBodyType.URL_VERIFICATION: self.handle_url_verification,
256
+
EventBodyType.EVENT_CALLBACK: self.handle_event_callback,
262
257
}
263
258
response = handlers.get(self.event_body.type, self.not_found)()
264
259
return response
265
260
266
-
def not_found(self):
267
-
...
261
+
def not_found(self) -> list:
262
+
return []
268
263
269
-
def handle_url_verification(self):
264
+
def handle_url_verification(self) -> dict:
265
+
"""Handles the endpoint verification."""
270
266
return {"challenge": self.event_body.challenge}
271
267
272
-
def handle_event_callback(self):
268
+
def handle_event_callback(self) -> list:
273
269
self.slack.publish_view(user_id=self.user.slack_user_id, view=HomeView())
274
270
return []
-2
kefi/routers/messages.py
-2
kefi/routers/messages.py