78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
from datetime import datetime
|
|
|
|
from clifinance.application.abstractions.protocols.driver import FileDriver
|
|
from clifinance.application.abstractions.protocols.session import Session
|
|
from clifinance.domain.expense.model import Expense, ExpenseType
|
|
|
|
|
|
class JsonFileSession(Session):
|
|
def __init__(self, driver: FileDriver) -> None:
|
|
self.driver = driver
|
|
self.data: list[Expense] = driver.read()
|
|
self.__dirty_new: list[Expense] = list()
|
|
self.__dirty_updated: list[Expense] = list()
|
|
self.__dirty_deleted: list[Expense] = list()
|
|
|
|
def commit(self) -> None:
|
|
for expense in self.__dirty_new:
|
|
self.data.append(expense)
|
|
|
|
for expense in self.__dirty_updated:
|
|
for item in self.data:
|
|
if item.id == expense.id:
|
|
item = expense
|
|
break
|
|
|
|
for expense in self.__dirty_deleted:
|
|
for item in self.data:
|
|
if item.id == expense.id:
|
|
del item
|
|
break
|
|
self.__dirty_new.clear()
|
|
self.__dirty_updated.clear()
|
|
self.__dirty_deleted.clear()
|
|
|
|
self.driver.write(self.data)
|
|
|
|
def rollback(self) -> None:
|
|
self.data = self.driver.read()
|
|
|
|
def add(self, expense: Expense) -> None:
|
|
# if not self.data:
|
|
# last_id = -1
|
|
# else:
|
|
# last_id = max({x.id for x in self.data if x.id is not None})
|
|
#
|
|
self.__dirty_new.append(expense)
|
|
|
|
def update(self, expense: Expense) -> None:
|
|
self.__dirty_updated.append(expense)
|
|
|
|
def delete(self, expense: Expense) -> None:
|
|
self.__dirty_deleted.append(expense)
|
|
|
|
def get(
|
|
self,
|
|
expense_id: int | None = None,
|
|
type: ExpenseType | None = None,
|
|
date: datetime | None = None,
|
|
amount: int | None = None,
|
|
) -> list[Expense]:
|
|
result: list[Expense] = list()
|
|
for expense in self.data:
|
|
if expense_id is not None and expense.id != expense_id:
|
|
continue
|
|
if type is not None and expense.type != type:
|
|
continue
|
|
if date is not None and expense.date != date:
|
|
continue
|
|
if amount is not None and expense.amount != amount:
|
|
continue
|
|
|
|
result.append(expense)
|
|
|
|
return result
|
|
|
|
def get_all(self) -> list[Expense]:
|
|
return self.data
|