Python backend for a Slack's kudos plugin.
feat(home): add evets route
+43
-9
+13
kefi/dependencies.py
+13
kefi/dependencies.py
···
1
1
from fastapi import Form
2
+
from pydantic import BaseModel
2
3
from sqlmodel import Session
3
4
4
5
from kefi.models.database import engine
···
7
8
class InteractionParams:
8
9
def __init__(self, payload: str = Form(...)):
9
10
self.payload = payload
11
+
12
+
13
+
class EventType(BaseModel):
14
+
type: str
15
+
user: str
16
+
17
+
18
+
class EventBody(BaseModel):
19
+
token: str
20
+
challenge: str | None
21
+
type: str
22
+
event: EventType
10
23
11
24
12
25
class SlashCommandParams:
+2
-1
kefi/main.py
+2
-1
kefi/main.py
···
4
4
from kefi.models.core.helpers import create_users
5
5
from kefi.models.database import engine
6
6
from kefi.models.kudos.helpers import create_default_actions
7
-
from kefi.routers import commands, interactivity
7
+
from kefi.routers import commands, events, interactivity
8
8
9
9
app = FastAPI()
10
10
app.include_router(commands.router)
11
11
app.include_router(interactivity.router)
12
+
app.include_router(events.router)
12
13
13
14
14
15
@app.on_event("startup")
+20
kefi/routers/events.py
+20
kefi/routers/events.py
···
1
+
from fastapi import APIRouter, Depends
2
+
from sqlmodel import Session
3
+
4
+
from kefi.dependencies import EventBody, InteractionParams, get_session
5
+
from kefi.routers.helpers import InteractionHandler
6
+
from kefi.routers.views import HomeView
7
+
from kefi.slack import Slack
8
+
9
+
router = APIRouter()
10
+
11
+
12
+
@router.post("/events/", tags=["events"])
13
+
def handle_events(
14
+
body: EventBody,
15
+
):
16
+
"""Calls the interaction handler and gets the response."""
17
+
print(body.event.type)
18
+
slack = Slack()
19
+
slack.publish_view(user_id=body.event.user, view=HomeView())
20
+
return {"challenge": body.challenge}
+1
kefi/routers/helpers.py
+1
kefi/routers/helpers.py
+4
-8
kefi/routers/views.py
+4
-8
kefi/routers/views.py
···
83
83
],
84
84
)
85
85
86
+
86
87
class HomeView(View):
87
88
def __init__(self, *args, **kwargs):
88
-
89
89
super().__init__(
90
90
type="home",
91
-
title="Bienvenido a Kefi",
92
-
close="Cerrar",
93
-
# TODO: not needed: submit=PlainTextObject(text="No asistiré"),
91
+
# TODO: not needed: submit=PlainTextObject(text="No asistiré"),
94
92
blocks=[
95
93
SectionBlock(
96
94
text=MarkdownTextObject(
···
99
97
),
100
98
DividerBlock(),
101
99
SectionBlock(
102
-
text=MarkdownTextObject(
103
-
text="This is so much fun!"
104
-
),
100
+
text=MarkdownTextObject(text="This is so much fun!"),
105
101
accessory=ImageElement(
106
102
image_url="https://api.slack.com/img/blocks/bkb_template_images/notifications.png",
107
103
alt_text="calendar thumbnail",
···
118
114
DividerBlock(),
119
115
SectionBlock(text=MarkdownTextObject(text="*¿Cambio de planes?*")),
120
116
],
121
-
)
117
+
)
+3
kefi/slack.py
+3
kefi/slack.py
···
56
56
def push_view(self, trigger_id: str, view: View):
57
57
self.client.views_push(trigger_id=trigger_id, view=view)
58
58
59
+
def publish_view(self, user_id: str, view: View):
60
+
self.client.views_publish(user_id=user_id, view=view)
61
+
59
62
def create_group_call(
60
63
self, url: str, users: Sequence[str], external_unique_id: str | None = None
61
64
):