151 lines
4 KiB
Python
151 lines
4 KiB
Python
from datetime import datetime
|
|
import logging
|
|
import sqlite3
|
|
from typing import Optional
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Transaction:
|
|
id: int
|
|
email_message_id: str
|
|
amount: str
|
|
card_ending_in: str
|
|
merchant: str
|
|
date: str
|
|
time: str
|
|
acknowledged: bool
|
|
ts_update: datetime
|
|
ts_insert: datetime
|
|
|
|
|
|
class TransactionManager:
|
|
def __init__(self, log: logging.Logger, db_conn: sqlite3.Connection):
|
|
self.__log = log
|
|
self.__db_conn = db_conn
|
|
|
|
def get_by_email_message_id(self, email_message_id: str) -> Optional[Transaction]:
|
|
result = self.__db_conn.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
email_message_id,
|
|
amount,
|
|
card_ending_in,
|
|
merchant,
|
|
date,
|
|
time,
|
|
acknowledged,
|
|
ts_update,
|
|
ts_insert
|
|
FROM "transaction"
|
|
WHERE email_message_id = ?
|
|
""",
|
|
(email_message_id,),
|
|
).fetchone()
|
|
|
|
if result is None:
|
|
return None
|
|
|
|
return Transaction(
|
|
id=result[0],
|
|
email_message_id=result[1],
|
|
amount=result[2],
|
|
card_ending_in=result[3],
|
|
merchant=result[4],
|
|
date=result[5],
|
|
time=result[6],
|
|
acknowledged=result[7],
|
|
ts_update=datetime.fromtimestamp(result[8]),
|
|
ts_insert=datetime.fromtimestamp(result[9]),
|
|
)
|
|
|
|
def acknowledge_transaction(self, transaction_id: int) -> Transaction:
|
|
row = self.__db_conn.execute(
|
|
"""
|
|
UPDATE "transaction"
|
|
SET acknowledged = TRUE, ts_update = strftime('%s')
|
|
WHERE id = ?
|
|
RETURNING
|
|
id,
|
|
email_message_id,
|
|
amount,
|
|
card_ending_in,
|
|
merchant,
|
|
date,
|
|
time,
|
|
acknowledged,
|
|
ts_update,
|
|
ts_insert
|
|
""",
|
|
(transaction_id,),
|
|
).fetchone()
|
|
|
|
return Transaction(
|
|
id=row[0],
|
|
email_message_id=row[1],
|
|
amount=row[2],
|
|
card_ending_in=row[3],
|
|
merchant=row[4],
|
|
date=row[5],
|
|
time=row[6],
|
|
acknowledged=row[7],
|
|
ts_update=datetime.fromtimestamp(row[8]),
|
|
ts_insert=datetime.fromtimestamp(row[9]),
|
|
)
|
|
|
|
|
|
def insert_transaction(self, transaction: Transaction) -> Transaction:
|
|
now = datetime.now()
|
|
result = self.__db_conn.execute(
|
|
"""
|
|
INSERT INTO "transaction" (
|
|
email_message_id,
|
|
amount,
|
|
card_ending_in,
|
|
merchant,
|
|
date,
|
|
time,
|
|
acknowledged,
|
|
ts_update,
|
|
ts_insert
|
|
) VALUES (
|
|
?,
|
|
?,
|
|
?,
|
|
?,
|
|
?,
|
|
?,
|
|
?,
|
|
?,
|
|
?
|
|
) RETURNING id
|
|
""",
|
|
(
|
|
transaction.email_message_id,
|
|
transaction.amount,
|
|
transaction.card_ending_in,
|
|
transaction.merchant,
|
|
transaction.date,
|
|
transaction.time,
|
|
transaction.acknowledged,
|
|
int(now.strftime("%s")),
|
|
int(now.strftime("%s")),
|
|
),
|
|
).fetchone()
|
|
|
|
self.__db_conn.commit()
|
|
|
|
return Transaction(
|
|
id=result[0],
|
|
email_message_id=transaction.email_message_id,
|
|
amount=transaction.amount,
|
|
card_ending_in=transaction.card_ending_in,
|
|
merchant=transaction.merchant,
|
|
date=transaction.date,
|
|
time=transaction.time,
|
|
acknowledged=transaction.acknowledged,
|
|
ts_update=now,
|
|
ts_insert=now,
|
|
)
|