Python backend for a Slack's kudos plugin.
feat: async task to create plazas
+40
-5
+7
kefi/models/database.py
+7
kefi/models/database.py
···
1
1
import datetime
2
2
from typing import Optional
3
3
4
+
import pytz
4
5
from sqlalchemy import Column, String
5
6
from sqlmodel import Field, Relationship, SQLModel, UniqueConstraint, create_engine
6
7
···
101
102
unique=True, index=True
102
103
) # When the meetings are going to be created
103
104
attendances: list["Attendance"] = Relationship(back_populates="plaza")
105
+
106
+
def local_date(self) -> datetime.datetime:
107
+
"""Gets the local date, using default Europe/Madrid timezone."""
108
+
return self.date.replace(tzinfo=pytz.utc).astimezone(
109
+
pytz.timezone("Europe/Madrid")
110
+
)
104
111
105
112
106
113
class Attendance(SQLModel, table=True):
+9
-1
kefi/models/plazas/helpers.py
+9
-1
kefi/models/plazas/helpers.py
···
33
33
34
34
def get_or_create_current_plaza(session: Session) -> Plaza:
35
35
"""Gets the current available plaza."""
36
-
now = datetime.datetime.now()
36
+
now = datetime.datetime.now(tz=pytz.utc)
37
37
query = select(Plaza).filter(Plaza.date >= now).order_by("date")
38
38
results = session.exec(query)
39
39
first = results.first()
···
121
121
external_unique_id=meet_id,
122
122
)
123
123
slack.notify_call(users=slack_users, call_id=call["id"])
124
+
125
+
126
+
def select_current_plaza(session: Session) -> Plaza | None:
127
+
"""Gets the curren plaza that have to ve executed."""
128
+
now = datetime.datetime.now(tz=pytz.utc).replace(second=0, microsecond=0)
129
+
query = select(Plaza).filter(Plaza.date == now).order_by("date")
130
+
results = session.exec(query)
131
+
return results.first()
+4
-2
kefi/routers/views.py
+4
-2
kefi/routers/views.py
···
16
16
def __init__(self, *args, **kwargs):
17
17
session = kwargs.get("session")
18
18
next_plaza = get_or_create_current_plaza(session=session)
19
+
date = next_plaza.local_date()
19
20
super().__init__(
20
21
callback_id=ViewType.JOIN_MEET,
21
22
type="modal",
···
31
32
DividerBlock(),
32
33
SectionBlock(
33
34
text=MarkdownTextObject(
34
-
text=f"*Próximo encuentro*\n{next_plaza.date.strftime('%A, %-d %B')}\n{next_plaza.date.strftime('%H:%M')}\n"
35
+
text=f"*Próximo encuentro*\n{date.strftime('%A, %-d %B')}\n{date.strftime('%H:%M')}\n"
35
36
),
36
37
accessory=ImageElement(
37
38
image_url="https://api.slack.com/img/blocks/bkb_template_images/notifications.png",
···
46
47
def __init__(self, *args, **kwargs):
47
48
session = kwargs.get("session")
48
49
next_plaza = get_or_create_current_plaza(session=session)
50
+
date = next_plaza.local_date()
49
51
super().__init__(
50
52
callback_id=ViewType.LEAVE_MEET,
51
53
type="modal",
···
61
63
DividerBlock(),
62
64
SectionBlock(
63
65
text=MarkdownTextObject(
64
-
text=f"*Kefi Plaza*\n{next_plaza.date.strftime('%A, %-d %B')}\n{next_plaza.date.strftime('%H:%M')}\n"
66
+
text=f"*Kefi Plaza*\n{date.strftime('%A, %-d %B')}\n{date.strftime('%H:%M')}\n"
65
67
),
66
68
accessory=ImageElement(
67
69
image_url="https://api.slack.com/img/blocks/bkb_template_images/notifications.png",
+5
-1
kefi/tasks/main.py
+5
-1
kefi/tasks/main.py
···
4
4
5
5
from kefi.config import settings
6
6
from kefi.models.database import engine
7
+
from kefi.tasks.plazas import runs_plaza
7
8
from kefi.tasks.wallets import recharge_wallets_task
8
9
9
10
···
18
19
19
20
class WorkerSettings:
20
21
redis_settings = RedisSettings(host=settings.REDIS_HOST)
21
-
cron_jobs = [cron(recharge_wallets_task, day=1, hour=0, minute=0)] # type: ignore
22
+
cron_jobs = [
23
+
cron(recharge_wallets_task, day=1, hour=0, minute=0), # type: ignore
24
+
cron(runs_plaza, minute=0), # type: ignore
25
+
]
22
26
on_startup = startup
23
27
on_shutdown = shutdown
+11
kefi/tasks/plazas.py
+11
kefi/tasks/plazas.py
···
1
+
from kefi.models.plazas.helpers import notify_plaza, select_current_plaza
2
+
3
+
4
+
async def runs_plaza(ctx):
5
+
"""Task to runs the plaza, that means, create the groups and send the call."""
6
+
session = ctx["session"]
7
+
plaza = select_current_plaza(session=session)
8
+
if plaza:
9
+
notify_plaza(session=session, plaza=plaza)
10
+
# In case the helper saves info
11
+
ctx["session"].commit()
+4
-1
kefi/tests/test_models.py
+4
-1
kefi/tests/test_models.py
···
1
1
import datetime
2
2
3
+
import pytz
3
4
from sqlmodel import Session
4
5
5
6
from kefi.config import settings
···
104
105
def test_get_or_create_current_plaza_with_future_and_past(session: Session):
105
106
first_plaza = get_or_create_current_plaza(session=session)
106
107
session.add(Plaza(date=first_plaza.date + datetime.timedelta(days=3)))
107
-
session.add(Plaza(date=datetime.datetime.now() - datetime.timedelta(days=3)))
108
+
session.add(
109
+
Plaza(date=datetime.datetime.now(tz=pytz.utc) - datetime.timedelta(days=3))
110
+
)
108
111
assert isinstance(first_plaza, Plaza)
109
112
second_plaza = get_or_create_current_plaza(session=session)
110
113
assert isinstance(second_plaza, Plaza)