main structure

.gitignore

@@ -1,3 +1,4 @@
+old_data/
 
 ### Python template
 # Byte-compiled / optimized / DLL files

src/clifinance/application/abstractions/protocols/driver.py

@@ -0,0 +1,14 @@
+from abc import abstractmethod
+from typing import Protocol
+
+from clifinance.application.dto.expense import ExpenseDTO
+
+
+class FileDriver(Protocol):
+    @abstractmethod
+    def read(self):
+        raise NotImplementedError
+
+    @abstractmethod
+    def write(self, data: list[ExpenseDTO]):
+        raise NotImplementedError

src/clifinance/application/abstractions/protocols/gateway.py

@@ -0,0 +1,36 @@
+from abc import abstractmethod
+from datetime import datetime
+from typing import Protocol
+
+from clifinance.application.dto.balance import Balance
+from clifinance.application.dto.expense import ExpenseDTO, ExpenseTypeDTO
+
+
+class DataGateway(Protocol):
+    @abstractmethod
+    def get_balance(self) -> Balance:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_all_expenses(self) -> list[ExpenseDTO]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def add_expense(self, exp: ExpenseDTO) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def update_expense(self, expense: ExpenseDTO) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_all_expenses_by_type(self, type: ExpenseTypeDTO) -> list[ExpenseDTO]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_all_expenses_by_date(self, date: datetime) -> list[ExpenseDTO]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_all_expenses_by_price(self, price: float) -> list[ExpenseDTO]:
+        raise NotImplementedError

src/clifinance/application/abstractions/protocols/session.py

@@ -0,0 +1,41 @@
+from abc import abstractmethod
+from datetime import datetime
+from typing import Protocol
+
+from clifinance.application.dto.expense import ExpenseDTO, ExpenseTypeDTO
+
+
+class FileSession(Protocol):
+    @abstractmethod
+    def commit(self) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def rollback(self) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def add(self, expense: ExpenseDTO) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def update(self, expense: ExpenseDTO) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def delete(self, expense: ExpenseDTO) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get(
+        self,
+        expense_id: int | None,
+        type: ExpenseTypeDTO | None,
+        date: datetime | None,
+        amount: int | None,
+    ) -> list[ExpenseDTO]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_all(self) -> list[ExpenseDTO]:
+        raise NotImplementedError

src/clifinance/application/dto/balance.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Balance:
+    balance: float
+    imcome_balance: float
+    expense_balance: float

src/clifinance/application/dto/expense.py

@@ -0,0 +1,36 @@
+from dataclasses import dataclass
+from datetime import datetime
+from enum import Enum
+
+
+class ExpenseTypeDTO(Enum):
+    INCOME = "Income"
+    EXPENSE = "Expense"
+
+
+@dataclass(frozen=True)
+class ExpenseDTO:
+    id: int | None
+    type: ExpenseTypeDTO
+    date: datetime
+    amount: float
+    description: str
+
+    def to_json(self) -> dict:
+        return {
+            "id": self.id,
+            "type": self.type.value,
+            "date": self.date.isoformat(),
+            "amount": self.amount,
+            "description": self.description,
+        }
+
+    @staticmethod
+    def from_json(data: dict) -> "ExpenseDTO":
+        return ExpenseDTO(
+            id=data["id"],
+            type=(ExpenseTypeDTO.INCOME if data["type"] == "Income" else ExpenseTypeDTO.EXPENSE),
+            date=datetime.fromisoformat(data["date"]),
+            amount=data["amount"],
+            description=data["description"],
+        )

src/clifinance/application/usecases/add_expense.py

@@ -0,0 +1,23 @@
+import argparse
+
+from clifinance.application.abstractions.protocols.gateway import DataGateway
+from clifinance.application.dto.expense import ExpenseDTO, ExpenseTypeDTO
+
+
+class AddExpense:
+    def __init__(
+        self,
+        gateway: DataGateway,
+    ) -> None:
+        self.__gateway = gateway
+
+    def __call__(self, args: argparse.Namespace) -> None:
+        self.__gateway.add_expense(
+            exp=ExpenseDTO(
+                id=None,
+                type=(ExpenseTypeDTO.INCOME if args.type == "income" else ExpenseTypeDTO.EXPENSE),
+                date=args.date,
+                amount=args.amount,
+                description=args.description,
+            )
+        )

src/clifinance/application/usecases/get_balance.py

@@ -0,0 +1,14 @@
+from clifinance.application.abstractions.protocols.gateway import DataGateway
+from clifinance.application.dto.balance import Balance
+
+
+class GetBalance:
+    def __init__(
+        self,
+        gateway: DataGateway,
+    ) -> None:
+        self.__gateway = gateway
+
+    def __call__(self) -> Balance:
+
+        return self.__gateway.get_balance()

src/clifinance/infrastructure/json_driver.py

@@ -0,0 +1,26 @@
+import json
+
+from clifinance.application.dto.expense import ExpenseDTO
+
+
+class JsonFileDriver:
+    def read(self):
+        try:
+            with open("./app_data.json") as f:
+                data = json.load(f)
+
+            return self.__data_from_json(data)
+        except FileNotFoundError:
+            return list()
+        except json.JSONDecodeError:
+            return list()
+
+    def write(self, data: list[ExpenseDTO]):
+        with open("./app_data.json", "w") as f:
+            json.dump(self.__data_to_json(data), f)
+
+    def __data_to_json(self, data: list[ExpenseDTO]):
+        return [x.to_json() for x in data]
+
+    def __data_from_json(self, data: list[dict]):
+        return [ExpenseDTO.from_json(x) for x in data]

src/clifinance/infrastructure/json_gateway.py

@@ -0,0 +1,41 @@
+from datetime import datetime
+
+from clifinance.application.abstractions.protocols.gateway import DataGateway
+from clifinance.application.abstractions.protocols.session import FileSession
+from clifinance.application.dto.balance import Balance
+from clifinance.application.dto.expense import ExpenseDTO, ExpenseTypeDTO
+
+
+class JsonDataGateway(DataGateway):
+    def __init__(self, session: FileSession) -> None:
+        self.session = session
+
+    def get_balance(self) -> Balance:
+        data = self.session.get_all()
+        balance = sum(x.amount for x in data)
+        income = sum(x.amount for x in data if x.type == ExpenseTypeDTO.INCOME)
+        expense = sum(x.amount for x in data if x.type == ExpenseTypeDTO.EXPENSE)
+        return Balance(
+            balance=round(balance, 2),
+            imcome_balance=round(income, 2),
+            expense_balance=round(expense, 2),
+        )
+
+    def get_all_expenses(self) -> list[ExpenseDTO]:
+        return self.session.get_all()
+
+    def add_expense(self, exp: ExpenseDTO) -> None:
+        self.session.add(exp)
+        self.session.commit()
+
+    def update_expense(self, expense: ExpenseDTO) -> None:
+        self.session.update(expense)
+
+    def get_all_expenses_by_type(self, type: ExpenseTypeDTO) -> list[ExpenseDTO]:
+        return [x for x in self.session.get_all() if x.type == type]
+
+    def get_all_expenses_by_date(self, date: datetime) -> list[ExpenseDTO]:
+        return [x for x in self.session.get_all() if x.date == date]
+
+    def get_all_expenses_by_price(self, price: float) -> list[ExpenseDTO]:
+        return [x for x in self.session.get_all() if x.amount == price]

src/clifinance/infrastructure/json_session.py

@@ -0,0 +1,85 @@
+from datetime import datetime
+
+from clifinance.application.abstractions.protocols.driver import FileDriver
+from clifinance.application.abstractions.protocols.session import FileSession
+from clifinance.application.dto.expense import ExpenseDTO, ExpenseTypeDTO
+
+
+class JsonFileSession(FileSession):
+    def __init__(self, driver: FileDriver) -> None:
+        self.driver = driver
+        self.data: list[ExpenseDTO] = driver.read()
+        self.__dirty_new: list[ExpenseDTO] = list()
+        self.__dirty_updated: list[ExpenseDTO] = list()
+        self.__dirty_deleted: list[ExpenseDTO] = 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: ExpenseDTO) -> 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(
+            ExpenseDTO(
+                id=last_id + 1,
+                type=expense.type,
+                date=expense.date,
+                amount=expense.amount,
+                description=expense.description,
+            )
+        )
+
+    def update(self, expense: ExpenseDTO) -> None:
+        self.__dirty_updated.append(expense)
+
+    def delete(self, expense: ExpenseDTO) -> None:
+        self.__dirty_deleted.append(expense)
+
+    def get(
+        self,
+        expense_id: int | None = None,
+        type: ExpenseTypeDTO | None = None,
+        date: datetime | None = None,
+        amount: int | None = None,
+    ) -> list[ExpenseDTO]:
+        result: list[ExpenseDTO] = 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[ExpenseDTO]:
+        return self.data

src/clifinance/main/app.py

@@ -0,0 +1,13 @@
+from clifinance.main.container import Container
+from clifinance.presentation.cli.parsers import provide_parser
+
+
+def main() -> None:
+    parser = provide_parser()
+
+    args = parser.parse_args()
+    args.func(args=args, ioc=Container())
+
+
+if __name__ == "__main__":
+    main()

src/clifinance/main/container.py

@@ -0,0 +1,48 @@
+from collections.abc import Callable, Generator
+from contextlib import contextmanager
+from functools import partial
+from inspect import signature
+from typing import ParamSpec, TypeVar
+
+from clifinance.application.abstractions.protocols.driver import FileDriver
+from clifinance.application.abstractions.protocols.gateway import DataGateway
+from clifinance.application.abstractions.protocols.session import FileSession
+from clifinance.application.usecases.add_expense import AddExpense
+from clifinance.application.usecases.get_balance import GetBalance
+from clifinance.infrastructure.json_driver import JsonFileDriver
+from clifinance.infrastructure.json_gateway import JsonDataGateway
+from clifinance.infrastructure.json_session import JsonFileSession
+
+
+class Container:
+
+    @contextmanager
+    def provide_get_balance(self) -> Generator[GetBalance, None, None]:
+        yield GetBalance(gateway=self.json_gateway())
+
+    @contextmanager
+    def provide_add_expense(self) -> Generator[AddExpense, None, None]:
+        yield AddExpense(gateway=self.json_gateway())
+
+    def json_gateway(self) -> DataGateway:
+        return JsonDataGateway(session=self.get_session())
+
+    def get_session(self) -> FileSession:
+        return JsonFileSession(driver=self.get_driver())
+
+    def get_driver(self) -> FileDriver:
+        return JsonFileDriver()
+
+
+P = ParamSpec("P")
+T = TypeVar("T")
+
+
+def provide_ioc(func: Callable[P, T], ioc: Container) -> partial[T] | None:
+    func_params = signature(func).parameters
+
+    for name, param in func_params.items():
+        if (param.annotation is Container) or (name == "ioc"):
+            return partial(func, **{name: ioc})
+
+    return None

src/clifinance/presentation/cli/commands.py

@@ -0,0 +1,27 @@
+import argparse
+
+from clifinance.main.container import Container
+
+
+def get_balance(args: argparse.Namespace, ioc: Container) -> None:
+    with ioc.provide_get_balance() as usecase:
+        print(usecase())
+
+
+def select_expenses(args: argparse.Namespace, ioc: Container) -> None:
+    print(args)
+
+
+def add_expense(args: argparse.Namespace, ioc: Container) -> None:
+    if not all(args.__dict__.values()):
+        raise argparse.ArgumentError(
+            argument=None,
+            message="All arguments are required: -t, -d, -a, -D",
+        )
+    if args.type == "income" and args.amount < 0:
+        raise argparse.ArgumentError(argument=None, message="An income cannot have a negative amount.")
+    if args.type == "expense" and args.amount > 0:
+        raise argparse.ArgumentError(argument=None, message="An expense cannot have a positive amount.")
+
+    with ioc.provide_add_expense() as usecase:
+        usecase(args=args)

src/clifinance/presentation/cli/parsers.py

@@ -0,0 +1,83 @@
+import argparse
+import datetime
+
+from clifinance.presentation.cli.commands import (
+    add_expense,
+    get_balance,
+    select_expenses,
+)
+
+
+def get_balance_parser(
+    subparser: argparse._SubParsersAction,
+) -> argparse.ArgumentParser:
+    balance_parser = subparser.add_parser(name="balance", help="Get balance")
+
+    return balance_parser
+
+
+def get_select_parser(subparser: argparse._SubParsersAction) -> argparse.ArgumentParser:
+    select_parser = subparser.add_parser(name="select", help="Get expenses by filter")
+    select_parser.add_argument(
+        "-d",
+        "--date",
+        type=datetime.date.fromisoformat,
+        help="Select expenses by date",
+    )
+    select_parser.add_argument(
+        "-p",
+        "--price",
+        type=float,
+        help="Select expenses by price",
+    )
+    select_parser.add_argument(
+        "-t",
+        "--type",
+        type=str,
+        help="Select expenses by type. Available values: income, expense",
+    )
+    return select_parser
+
+
+def get_add_parser(subparser: argparse._SubParsersAction) -> argparse.ArgumentParser:
+    add_parser = subparser.add_parser(name="add", help="Add new expense")
+    add_parser.add_argument(
+        "-t",
+        "--type",
+        type=str,
+        help="Expense type. Available values: income, expense",
+    )
+    add_parser.add_argument(
+        "-d",
+        "--date",
+        type=datetime.datetime.fromisoformat,
+        help="Expense date",
+    )
+    add_parser.add_argument(
+        "-a",
+        "--amount",
+        type=float,
+        help="Expense price",
+    )
+    add_parser.add_argument(
+        "-D",
+        "--description",
+        type=str,
+        help="Expense description",
+    )
+    return add_parser
+
+
+def provide_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Cli finance app")
+    subparser = parser.add_subparsers(title="Availeble commands", dest="command")
+
+    balance_parser = get_balance_parser(subparser=subparser)
+    balance_parser.set_defaults(func=get_balance)
+
+    select_parser = get_select_parser(subparser=subparser)
+    select_parser.set_defaults(func=select_expenses)
+
+    add_parser = get_add_parser(subparser=subparser)
+    add_parser.set_defaults(func=add_expense)
+    return parser