1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
|
from typing import Any
from psycopg import sql
from jb.managers import PostgresManager
from jb.models.bonus import Bonus
class BonusManager(PostgresManager):
def create(self, bonus: Bonus) -> None:
assert bonus.id is None
data = bonus.to_postgres()
query = sql.SQL("""
INSERT INTO mtwerk_bonus
(payout_event_id, amt_worker_id, amount, grant_time, assignment_id, reason)
VALUES (
%(payout_event_id)s,
%(amt_worker_id)s,
%(amount)s,
%(grant_time)s,
(
SELECT id
FROM mtwerk_assignment
WHERE amt_assignment_id = %(amt_assignment_id)s
LIMIT 1
),
%(reason)s
)
RETURNING id, assignment_id;
""")
with self.pg_config.make_connection() as conn, conn.cursor() as c:
c.execute(query, data)
res = c.fetchone()
conn.commit()
bonus.id = res["id"] # type: ignore
bonus.assignment_id = res["assignment_id"] # type: ignore
def filter(self, amt_assignment_id: str) -> list[Bonus]:
res: list[Any] = self.pg_config.execute_sql_query(
"""
SELECT mb.*, ma.amt_assignment_id
FROM mtwerk_bonus mb
JOIN mtwerk_assignment ma ON ma.id = mb.assignment_id
WHERE amt_assignment_id = %(amt_assignment_id)s;
""",
params={"amt_assignment_id": amt_assignment_id},
)
return [Bonus.from_postgres(data=x) for x in res]
|
