Python backend for a Slack's kudos plugin.
chore: using python 3.10 type notations
+83
-89
+19
-21
kefi/models/database.py
+19
-21
kefi/models/database.py
···
1
-
from typing import List, 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] = ""
19
-
transactions: List["Transaction"] = Relationship(
16
+
slack_username: str | None = ""
17
+
transactions: list["Transaction"] = Relationship(
20
18
back_populates="user",
21
19
sa_relationship_kwargs=dict(foreign_keys="[Transaction.user_id]"),
22
20
)
23
-
transactions_sent: List["Transaction"] = Relationship(
21
+
transactions_sent: list["Transaction"] = Relationship(
24
22
back_populates="sender",
25
23
sa_relationship_kwargs=dict(foreign_keys="[Transaction.sender_id]"),
26
24
)
27
-
transactions_received: List["Transaction"] = Relationship(
25
+
transactions_received: list["Transaction"] = Relationship(
28
26
back_populates="receiver",
29
27
sa_relationship_kwargs=dict(foreign_keys="[Transaction.receiver_id]"),
30
28
)
···
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
-
transactions: List["Transaction"] = Relationship(back_populates="action")
42
+
transactions: list["Transaction"] = Relationship(back_populates="action")
45
43
# Responses data
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
)
+25
-25
kefi/models/helpers.py
+25
-25
kefi/models/helpers.py
···
1
-
from typing import Dict, List, Optional, Tuple, Union
2
-
3
1
from slack_sdk import WebClient
4
2
from sqlmodel import Session, func, or_, select
5
3
···
12
10
13
11
14
12
def get_or_create_from_command(
15
-
command: SlashCommandParams, session: Session
16
-
) -> Tuple[User, bool]:
13
+
command: "SlashCommandParams", session: "Session"
14
+
) -> tuple["User", bool]:
17
15
"""Obtains the balance of the user."""
18
16
query = select(User).filter(User.slack_user_id == command.user_id)
19
17
user = session.exec(query).one_or_none()
···
27
25
return (user, created)
28
26
29
27
30
-
def available_balance(user: User, session: Session) -> int:
31
-
"""Obtains the availabe balance of the user, that means the balance not spend from
28
+
def available_balance(user: "User", session: "Session") -> int:
29
+
"""Obtains the available balance of the user, that means the balance not spend from
32
30
the monthly received.
33
31
"""
34
32
query = select(func.sum(Transaction.amount)).filter( # type: ignore
···
38
36
return session.exec(query).one() or 0
39
37
40
38
41
-
def received_balance(user: User, session: Session) -> int:
39
+
def received_balance(user: "User", session: "Session") -> int:
42
40
"""Obtains the received balance of the user, that means the balance sent to the
43
41
user.
44
42
"""
···
49
47
50
48
51
49
def send_action(
52
-
sender: User, action: Action, receiver: User, message: str, session: Session
53
-
) -> List[Transaction]:
50
+
sender: "User", action: "Action", receiver: "User", message: str, session: "Session"
51
+
) -> list["Transaction"]:
54
52
"""Creates the transaction needed to send an action from sender to receiver."""
55
53
# Checks sender wallet
56
54
balance = available_balance(user=sender, session=session)
···
79
77
80
78
81
79
def notify_receiver_user_chat_action(
82
-
action: Action, sender: User, receiver: User, message: str, channel_id: str
80
+
action: "Action", sender: "User", receiver: "User", message: str, channel_id: str
83
81
):
84
82
slack = Slack()
85
83
action_response = ActionResponse(
···
104
102
)
105
103
106
104
107
-
def send_admin_amount(receiver: User, amount: int, session: Session) -> Transaction:
105
+
def send_admin_amount(
106
+
receiver: "User", amount: int, session: "Session"
107
+
) -> "Transaction":
108
108
"""Creates the transaction needed to send an amount to the receiver."""
109
109
receiver_transaction = Transaction(amount=amount, user=receiver)
110
110
session.add(receiver_transaction)
111
111
return receiver_transaction
112
112
113
113
114
-
def recharge_wallets(amount: int, session: Session):
114
+
def recharge_wallets(amount: int, session: "Session") -> None:
115
115
"""Recharge all the wallet with the given amount."""
116
116
query = select(User)
117
117
users = session.exec(query).all()
118
118
for user in users:
119
-
transaction = Transaction(amount=amount, user=user) # type: ignore
119
+
transaction = Transaction(amount=amount, user=user)
120
120
session.add(transaction)
121
121
122
122
123
-
def reset_wallets(session: Session):
123
+
def reset_wallets(session: "Session") -> None:
124
124
"""Reset the wallets of all the users."""
125
125
users = session.exec(select(User)).all()
126
126
for user in users:
127
127
balance = available_balance(user=user, session=session)
128
128
amount = settings.RECHARGE_KEFIS_AMOUNT - balance
129
-
transaction = Transaction(amount=amount, user=user) # type: ignore
129
+
transaction = Transaction(amount=amount, user=user)
130
130
session.add(transaction)
131
131
132
132
133
-
def notify_reset_wallet(session: Session) -> None:
133
+
def notify_reset_wallet(session: "Session") -> None:
134
134
"""Sends notifications after the wallet was reset."""
135
135
amount = settings.RECHARGE_KEFIS_AMOUNT
136
136
users = session.exec(select(User)).all()
···
140
140
slack.post_message_user(user.slack_user_id, blocks=response.render()["blocks"])
141
141
142
142
143
-
def create_default_actions(session: Session) -> None:
143
+
def create_default_actions(session: "Session") -> None:
144
144
"""Creates the default actions if they doesn't exists."""
145
-
actions: List[Dict[str, Union[int, str]]] = [
145
+
actions: list[dict[str, int | str]] = [
146
146
{
147
147
"keyword": Command.KUDOS,
148
148
"amount": 100,
···
172
172
},
173
173
]
174
174
for action_data in actions:
175
-
action: Optional[Action] = session.exec(
175
+
action: "Action" | None = session.exec(
176
176
select(Action).filter(Action.keyword == action_data["keyword"])
177
177
).one_or_none()
178
178
if not action:
···
183
183
session.add(action)
184
184
185
185
186
-
def find_user_by_slack_user_id(slack_user_id: str, session: Session) -> Optional[User]:
186
+
def find_user_by_slack_user_id(slack_user_id: str, session: "Session") -> "User" | None:
187
187
"""Gets the user using the Slack user id."""
188
188
query = select(User).filter(User.slack_user_id == slack_user_id)
189
189
return session.exec(query).one_or_none()
190
190
191
191
192
192
def find_user_by_slack_username(
193
-
slack_username: str, session: Session
194
-
) -> Optional[User]:
193
+
slack_username: str, session: "Session"
194
+
) -> "User" | None:
195
195
"""Gets the user using the Slack username."""
196
196
query = select(User).filter(User.slack_username == slack_username)
197
197
return session.exec(query).one_or_none()
198
198
199
199
200
-
def get_action(keyword: str, session: Session) -> Optional[Action]:
200
+
def get_action(keyword: str, session: "Session") -> "Action" | None:
201
201
query = select(Action).filter(Action.keyword == keyword)
202
202
return session.exec(query).one_or_none()
203
203
204
204
205
-
def create_users(session: Session) -> None:
205
+
def create_users(session: "Session") -> None:
206
206
"""Creates all the users in the team."""
207
207
# Gets the users
208
208
client = WebClient(token=settings.SLACK_BOT_TOKEN)
···
234
234
session.add(user)
235
235
236
236
237
-
def get_all_users(session: Session) -> List[User]:
237
+
def get_all_users(session: Session) -> list["User"]:
238
238
"""Gets the complete list of users."""
239
239
return session.exec(select(User)).all()
+11
-13
kefi/models/outputs.py
+11
-13
kefi/models/outputs.py
···
1
-
from typing import List, Optional, Union
2
-
3
1
from pydantic import BaseModel
4
2
5
3
···
33
31
"""Images accessories entries."""
34
32
35
33
type: str = "image"
36
-
image_url: Optional[str]
37
-
alt_text: Optional[str]
34
+
image_url: str | None
35
+
alt_text: str | None
38
36
39
37
40
38
class Block(BaseModel):
41
39
"""Base block entries."""
42
40
43
41
type: str
44
-
text: Optional[Text]
42
+
text: "Text" | None
45
43
46
44
47
45
class Section(Block):
48
46
"""Section blocks."""
49
47
50
48
type: str = "section"
51
-
fields: Optional[List[Text]]
52
-
accessory: Optional[Accessory]
49
+
fields: list["Text"] | None
50
+
accessory: "Accessory" | None
53
51
54
52
55
53
class Header(Block):
56
54
"""Header block."""
57
55
58
56
type: str = "header"
59
-
text: Text
57
+
text: "Text"
60
58
61
59
62
60
class Context(Block):
63
61
"""Context block"""
64
62
65
63
type: str = "context"
66
-
elements: List[Union[Text, Image]]
64
+
elements: list["Text" | "Image"]
67
65
68
66
69
67
class Response(BaseModel):
70
68
"""Main response for Slack commands."""
71
69
72
-
channel: Optional[str]
73
-
text: Optional[str]
74
-
response_type: Optional[str] = "ephemeral"
75
-
blocks: List[Block]
70
+
channel: str | None
71
+
text: str | None
72
+
response_type: str | None = "ephemeral"
73
+
blocks: list["Block"]
+11
-11
kefi/routers/helpers.py
+11
-11
kefi/routers/helpers.py
···
1
1
import re
2
-
from typing import Callable, Dict, List, Optional, Tuple
2
+
from collections.abc import Callable
3
3
4
4
from sqlmodel import Session
5
5
···
29
29
class CommandHandler:
30
30
"""Handle different commands."""
31
31
32
-
def __init__(self, command: SlashCommandParams, session: Session):
32
+
def __init__(self, command: "SlashCommandParams", session: "Session"): # type: ignore
33
33
self.session = session
34
34
self.command = command
35
35
self.user, _ = get_or_create_from_command(command=command, session=session)
36
36
37
-
def extract_keyword_params(self, text: str) -> Tuple[str, List[str]]:
37
+
def extract_keyword_params(self, text: str) -> tuple[str, list[str]]:
38
38
params = text.split()
39
39
return params[0], params[1:]
40
40
41
-
def extract_user_id(self, text: str) -> Optional[str]:
41
+
def extract_user_id(self, text: str) -> str | None:
42
42
found_user_id = re.search("<@([^\|]+)\|?(.+)?>", text)
43
43
return found_user_id.group(1) if found_user_id else None
44
44
45
-
def response(self) -> Dict:
45
+
def response(self) -> dict:
46
46
try:
47
47
keyword, params = self.extract_keyword_params(self.command.text)
48
48
except IndexError:
49
-
return SimpleResponse("No he entenido el comando").render()
50
-
handlers: Dict[str, Callable] = {
49
+
return SimpleResponse("No he entendido el comando").render()
50
+
handlers: dict[str, Callable] = {
51
51
Command.HELP: self.command_help,
52
52
Command.WALLET: self.command_wallet,
53
53
Command.KUDOS: self.command_action,
···
60
60
).render()
61
61
return response
62
62
63
-
def command_help(self, *args, **kwargs) -> SlackResponse:
63
+
def command_help(self, *args, **kwargs) -> "SlackResponse":
64
64
return HelpResponse()
65
65
66
66
def command_wallet(self, *args, **kwargs) -> SlackResponse:
···
68
68
received_amount = received_balance(user=self.user, session=self.session)
69
69
return WalletResponse(self.user, remaining_amount, received_amount)
70
70
71
-
def command_action(self, keyword: str, params: List[str]) -> SlackResponse:
71
+
def command_action(self, keyword: str, params: list[str]) -> "SlackResponse":
72
72
action = get_action(keyword=keyword, session=self.session)
73
73
if not action:
74
74
return SimpleResponse("No existe la acción asociada a este comando")
···
109
109
channel_id=self.command.channel_id,
110
110
)
111
111
112
-
def command_reward(self, keyword: str, params: List[str]) -> SlackResponse:
112
+
def command_reward(self, keyword: str, params: list[str]) -> "SlackResponse":
113
113
amount = int(params[0])
114
114
if not self.user.is_admin:
115
115
return SimpleResponse(
···
136
136
)
137
137
return SimpleResponse(f"Has enviado {amount} kefis")
138
138
139
-
def not_found(self, *args, **kwargs) -> SlackResponse:
139
+
def not_found(self, *args, **kwargs) -> "SlackResponse":
140
140
return NotFoundResponse()
+7
-9
kefi/routers/responses.py
+7
-9
kefi/routers/responses.py
···
1
-
from typing import Any, Dict, List, Optional
2
-
3
1
from kefi.models.database import Action, User
4
2
from kefi.models.outputs import (
5
3
Context,
···
15
13
class SlackResponse:
16
14
"""Base class for a Slack response."""
17
15
18
-
response: Optional[Response]
16
+
response: Response | None
19
17
20
-
def render(self) -> Dict:
18
+
def render(self) -> dict:
21
19
"""Render the response model, getting the dict.."""
22
20
return self.response.dict(exclude_none=True) if self.response else {}
23
21
···
62
60
class ActionResponse(SlackResponse):
63
61
def __init__(
64
62
self,
65
-
sender: User,
66
-
receiver: User,
67
-
action: Action,
63
+
sender: "User",
64
+
receiver: "User",
65
+
action: "Action",
68
66
message: str,
69
67
channel_id: str,
70
-
response_type: Optional[str] = "in_channel",
68
+
response_type: str | None = "in_channel",
71
69
) -> None:
72
70
super().__init__()
73
71
# Saves the data
···
82
80
83
81
def build_response(self) -> None:
84
82
"""Creates the response using the data in action and"""
85
-
blocks: List[Any] = []
83
+
blocks: list = []
86
84
blocks.append(
87
85
Header(
88
86
text=PlainText(
+4
-4
kefi/slack.py
+4
-4
kefi/slack.py
···
1
-
from typing import Optional, Sequence
1
+
from collections.abc import Sequence
2
2
3
3
from slack_sdk.web.client import WebClient
4
4
···
9
9
def __init__(self) -> None:
10
10
self.client = WebClient(token=settings.SLACK_BOT_TOKEN)
11
11
12
-
def get_users(self, cursor: Optional[str] = None, users=[]):
12
+
def get_users(self, cursor: str | None = None, users=[]):
13
13
result = self.client.users_list(team_id=settings.SLACK_TEAM_ID, cursor=cursor)
14
14
users = users + result["members"]
15
15
next_cursor = result["response_metadata"]["next_cursor"]
···
20
20
else:
21
21
return users
22
22
23
-
def get_user_chats(self, cursor: Optional[str] = None, user_chats=[]):
23
+
def get_user_chats(self, cursor: str | None = None, user_chats=[]):
24
24
result = self.client.conversations_list(
25
25
team_id=settings.SLACK_TEAM_ID, cursor=cursor, types=["im"]
26
26
)
···
40
40
return user_chats
41
41
42
42
def post_message_user(
43
-
self, user_id: str, blocks: Sequence[dict], text: Optional[str] = None
43
+
self, user_id: str, blocks: Sequence[dict], text: str | None = None
44
44
):
45
45
self.client.chat_postMessage(channel=user_id, blocks=blocks, text=text)
+4
-4
kefi/tests/test_commands.py
+4
-4
kefi/tests/test_commands.py
···
10
10
from kefi.tests.helpers import generate_command
11
11
12
12
13
-
def test_unknown_commad(session: Session, client: TestClient):
13
+
def test_unknown_command(session: Session, client: TestClient):
14
14
"""Test command /kefi unknown"""
15
15
16
16
response = client.post("/command/", data=generate_command(text="unknown"))
···
19
19
assert len(users) == 1
20
20
21
21
22
-
def test_empty_commad(client: TestClient):
22
+
def test_empty_command(client: TestClient):
23
23
"""Test command /kefi"""
24
24
25
25
response = client.post("/command/", data=generate_command(text=""))
26
26
assert response.status_code == 422
27
27
28
28
29
-
def test_help_commad(session: Session, client: TestClient):
29
+
def test_help_command(session: Session, client: TestClient):
30
30
"""Test command /kefi help"""
31
31
32
32
response = client.post("/command/", data=generate_command(text="help"))
···
38
38
assert len(data["blocks"]) == 4
39
39
40
40
41
-
def test_wallet_commad(session: Session, client: TestClient):
41
+
def test_wallet_command(session: Session, client: TestClient):
42
42
"""Test command /kefi wallet"""
43
43
44
44
user = User(slack_user_id="U1")
+2
-2
kefi/tests/test_slack.py
+2
-2
kefi/tests/test_slack.py
···
9
9
def test_get_users(mocker: MockerFixture):
10
10
slack = Slack()
11
11
users_list_mock = mocker.patch.object(WebClient, "users_list")
12
-
first_response = {
12
+
first_response: dict[str, list | dict] = {
13
13
"members": [{"name": "merkos"}, {"name": "rubon"}, {"name": "sondra"}],
14
14
"response_metadata": {"next_cursor": "next_cursor"},
15
15
}
16
-
second_response = {
16
+
second_response: dict[str, list | dict] = {
17
17
"members": [
18
18
{"name": "ivo"},
19
19
{"name": "marina"},