Python backend for a Slack's kudos plugin.
fix: typing
+40
-46
+2
-2
README.md
+2
-2
README.md
+15
-17
kefi/models/database.py
+15
-17
kefi/models/database.py
···
1
-
from typing import Optional
2
-
3
1
from sqlalchemy import Column, String
4
2
from sqlmodel import Field, Relationship, SQLModel, create_engine
5
3
···
11
9
class User(SQLModel, table=True): # type: ignore
12
10
"""A slack user."""
13
11
14
-
id: Optional[int] = Field(default=None, primary_key=True)
15
-
first_name: Optional[str] = ""
16
-
last_name: Optional[str] = ""
12
+
id: int | None = Field(default=None, primary_key=True)
13
+
first_name: str | None = ""
14
+
last_name: str | None = ""
17
15
slack_user_id: str = Field(sa_column=Column(String(100), unique=True, index=True))
18
-
slack_username: Optional[str] = ""
16
+
slack_username: str | None = ""
19
17
transactions: list["Transaction"] = Relationship(
20
18
back_populates="user",
21
19
sa_relationship_kwargs=dict(foreign_keys="[Transaction.user_id]"),
···
38
36
class Action(SQLModel, table=True): # type: ignore
39
37
"""Each action a user can perform to give kefis to another user."""
40
38
41
-
id: Optional[int] = Field(default=None, primary_key=True)
39
+
id: int | None = Field(default=None, primary_key=True)
42
40
keyword: str = Field(sa_column=Column(String(100), unique=True, index=True))
43
41
amount: int
44
42
transactions: list["Transaction"] = Relationship(back_populates="action")
···
46
44
header_template: str = ""
47
45
message_template: str = ""
48
46
context_template: str = ""
49
-
image: Optional[str]
50
-
text: Optional[str]
47
+
image: str | None
48
+
text: str | None
51
49
52
50
53
51
class Transaction(SQLModel, table=True): # type: ignore
···
55
53
another.
56
54
"""
57
55
58
-
id: Optional[int] = Field(default=None, primary_key=True)
56
+
id: int | None = Field(default=None, primary_key=True)
59
57
amount: int
60
58
user_id: int = Field(foreign_key="user.id")
61
59
user: User = Relationship(
62
60
back_populates="transactions",
63
61
sa_relationship_kwargs=dict(foreign_keys="[Transaction.user_id]"),
64
62
)
65
-
message: Optional[str] = Field(default=None)
63
+
message: str | None = Field(default=None)
66
64
67
65
# Action, the reference of the action used to send the transaction
68
-
action_id: Optional[int] = Field(default=None, foreign_key="action.id")
69
-
action: Optional["Action"] = Relationship(back_populates="transactions")
66
+
action_id: int | None = Field(default=None, foreign_key="action.id")
67
+
action: Action | None = Relationship(back_populates="transactions")
70
68
71
69
# Sender user, if not defined, is a system transaction
72
-
sender_id: Optional[int] = Field(default=None, foreign_key="user.id")
73
-
sender: Optional["User"] = Relationship(
70
+
sender_id: int | None = Field(default=None, foreign_key="user.id")
71
+
sender: User | None = Relationship(
74
72
back_populates="transactions_sent",
75
73
sa_relationship_kwargs=dict(foreign_keys="[Transaction.sender_id]"),
76
74
)
77
75
78
76
# Receiver user
79
-
receiver_id: Optional[int] = Field(default=None, foreign_key="user.id")
80
-
receiver: Optional["User"] = Relationship(
77
+
receiver_id: int | None = Field(default=None, foreign_key="user.id")
78
+
receiver: User | None = Relationship(
81
79
back_populates="transactions_received",
82
80
sa_relationship_kwargs=dict(foreign_keys="[Transaction.receiver_id]"),
83
81
)
+14
-18
kefi/models/helpers.py
+14
-18
kefi/models/helpers.py
···
10
10
11
11
12
12
def get_or_create_from_command(
13
-
command: "SlashCommandParams", session: "Session"
13
+
command: SlashCommandParams, session: Session
14
14
) -> tuple["User", bool]:
15
15
"""Obtains the balance of the user."""
16
16
query = select(User).filter(User.slack_user_id == command.user_id)
···
25
25
return (user, created)
26
26
27
27
28
-
def available_balance(user: "User", session: "Session") -> int:
28
+
def available_balance(user: User, session: Session) -> int:
29
29
"""Obtains the available balance of the user, that means the balance not spend from
30
30
the monthly received.
31
31
"""
···
36
36
return session.exec(query).one() or 0
37
37
38
38
39
-
def received_balance(user: "User", session: "Session") -> int:
39
+
def received_balance(user: User, session: Session) -> int:
40
40
"""Obtains the received balance of the user, that means the balance sent to the
41
41
user.
42
42
"""
···
47
47
48
48
49
49
def send_action(
50
-
sender: "User", action: "Action", receiver: "User", message: str, session: "Session"
50
+
sender: User, action: Action, receiver: User, message: str, session: Session
51
51
) -> list["Transaction"]:
52
52
"""Creates the transaction needed to send an action from sender to receiver."""
53
53
# Checks sender wallet
···
77
77
78
78
79
79
def notify_receiver_user_chat_action(
80
-
action: "Action", sender: "User", receiver: "User", message: str, channel_id: str
80
+
action: Action, sender: User, receiver: User, message: str, channel_id: str
81
81
):
82
82
slack = Slack()
83
83
action_response = ActionResponse(
···
102
102
)
103
103
104
104
105
-
def send_admin_amount(
106
-
receiver: "User", amount: int, session: "Session"
107
-
) -> "Transaction":
105
+
def send_admin_amount(receiver: User, amount: int, session: Session) -> "Transaction":
108
106
"""Creates the transaction needed to send an amount to the receiver."""
109
107
receiver_transaction = Transaction(amount=amount, user=receiver)
110
108
session.add(receiver_transaction)
111
109
return receiver_transaction
112
110
113
111
114
-
def recharge_wallets(amount: int, session: "Session") -> None:
112
+
def recharge_wallets(amount: int, session: Session) -> None:
115
113
"""Recharge all the wallet with the given amount."""
116
114
query = select(User)
117
115
users = session.exec(query).all()
···
130
128
session.add(transaction)
131
129
132
130
133
-
def notify_reset_wallet(session: "Session") -> None:
131
+
def notify_reset_wallet(session: Session) -> None:
134
132
"""Sends notifications after the wallet was reset."""
135
133
amount = settings.RECHARGE_KEFIS_AMOUNT
136
134
users = session.exec(select(User)).all()
···
140
138
slack.post_message_user(user.slack_user_id, blocks=response.render()["blocks"])
141
139
142
140
143
-
def create_default_actions(session: "Session") -> None:
141
+
def create_default_actions(session: Session) -> None:
144
142
"""Creates the default actions if they doesn't exists."""
145
143
actions: list[dict[str, int | str]] = [
146
144
{
···
172
170
},
173
171
]
174
172
for action_data in actions:
175
-
action: "Action" | None = session.exec(
173
+
action: Action | None = session.exec(
176
174
select(Action).filter(Action.keyword == action_data["keyword"])
177
175
).one_or_none()
178
176
if not action:
···
183
181
session.add(action)
184
182
185
183
186
-
def find_user_by_slack_user_id(slack_user_id: str, session: "Session") -> "User" | None:
184
+
def find_user_by_slack_user_id(slack_user_id: str, session: Session) -> User | None:
187
185
"""Gets the user using the Slack user id."""
188
186
query = select(User).filter(User.slack_user_id == slack_user_id)
189
187
return session.exec(query).one_or_none()
190
188
191
189
192
-
def find_user_by_slack_username(
193
-
slack_username: str, session: "Session"
194
-
) -> "User" | None:
190
+
def find_user_by_slack_username(slack_username: str, session: Session) -> User | None:
195
191
"""Gets the user using the Slack username."""
196
192
query = select(User).filter(User.slack_username == slack_username)
197
193
return session.exec(query).one_or_none()
198
194
199
195
200
-
def get_action(keyword: str, session: "Session") -> "Action" | None:
196
+
def get_action(keyword: str, session: Session) -> Action | None:
201
197
query = select(Action).filter(Action.keyword == keyword)
202
198
return session.exec(query).one_or_none()
203
199
204
200
205
-
def create_users(session: "Session") -> None:
201
+
def create_users(session: Session) -> None:
206
202
"""Creates all the users in the team."""
207
203
# Gets the users
208
204
client = WebClient(token=settings.SLACK_BOT_TOKEN)
+6
-6
kefi/models/outputs.py
+6
-6
kefi/models/outputs.py
···
39
39
"""Base block entries."""
40
40
41
41
type: str
42
-
text: "Text" | None
42
+
text: Text | None
43
43
44
44
45
45
class Section(Block):
46
46
"""Section blocks."""
47
47
48
48
type: str = "section"
49
-
fields: list["Text"] | None
50
-
accessory: "Accessory" | None
49
+
fields: list[Text] | None
50
+
accessory: Accessory | None
51
51
52
52
53
53
class Header(Block):
54
54
"""Header block."""
55
55
56
56
type: str = "header"
57
-
text: "Text"
57
+
text: Text
58
58
59
59
60
60
class Context(Block):
61
61
"""Context block"""
62
62
63
63
type: str = "context"
64
-
elements: list["Text" | "Image"]
64
+
elements: list[Text | Image]
65
65
66
66
67
67
class Response(BaseModel):
···
70
70
channel: str | None
71
71
text: str | None
72
72
response_type: str | None = "ephemeral"
73
-
blocks: list["Block"]
73
+
blocks: list[Block]
+3
-3
kefi/routers/responses.py
+3
-3
kefi/routers/responses.py