Refactor -> Unit of Work pattern

.github/workflows/docker.yaml

@@ -1,7 +1,8 @@
 ---
 name: Docker build and tests
 on:
-  push:
+  #  push:
+  pull_request:
     branches:
 env:
   DOCKER_IMAGE_NAME: parseltongist/bank_app
@@ -14,7 +15,7 @@ jobs:
         uses: actions/checkout@v4
       - name: Build Docker Image
         run: |
-          docker build -t test-tag .
+          docker build -t test-tag --build-arg="branch=$GITHUB_HEAD_REF" .
         working-directory: ${{ env.DOCKERFILE_DIRECTORY }}
       - name: TEST VIA DOCKER
         run: docker run test-tag

.github/workflows/pre-commit.yaml

@@ -1,26 +1,26 @@
-#name: pre-commit
-#on:
-#  pull_request:
-#  push:
-#env:
-#  DOCKER_IMAGE_NAME: parseltongist/bank_app
-#  DOCKERFILE_DIRECTORY: ./
-#jobs:
-#  pre-commit:
-#    name: pre-commit
-#    runs-on: ubuntu-latest
-#    steps:
-#      - uses: actions/checkout@v4
-#      - uses: actions/setup-python@v5
-#        with:
-#          python-version: 3.11
-#      - uses: pre-commit/[email protected]
-#  docker-build:
-#    runs-on: ubuntu-latest
-#    steps:
-#      - name: Checkout code
-#        uses: actions/checkout@v4
-#      - name: Build Docker Image
-#        run: |
-#          docker build -t $DOCKER_IMAGE_NAME:latest .
-#        working-directory: ${{ env.DOCKERFILE_DIRECTORY }}
+name: pre-commit
+on:
+  pull_request:
+  push:
+env:
+  DOCKER_IMAGE_NAME: parseltongist/bank_app
+  DOCKERFILE_DIRECTORY: ./
+jobs:
+  pre-commit:
+    name: pre-commit
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: 3.11
+      - uses: pre-commit/[email protected]
+  docker-build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+      - name: Build Docker Image
+        run: |
+          docker build -t $DOCKER_IMAGE_NAME:latest .
+        working-directory: ${{ env.DOCKERFILE_DIRECTORY }}

.github/workflows/run_tests.yaml

@@ -1,3 +1,4 @@
+---
 #---
 #name: Run unit tests
 #on:

.pre-commit-config.yaml

@@ -1,19 +1,15 @@
 ---
 default_language_version:
-  python: python3
+  python: python3.11
   node: "14.18.0"
 repos:
   - repo: https://github.com/pre-commit/mirrors-prettier
     rev: v2.4.1
     hooks:
       - id: prettier
-        name: prettier (with plugin-xml)
+        name: prettier
         additional_dependencies:
           - "[email protected]"
-          - "@prettier/[email protected]"
-        args:
-          - --plugin=@prettier/plugin-xml
-          - --xml-self-closing-space=false
         files: \.(css|htm|html|js|json|jsx|less|md|scss|toml|ts|xml|yaml|yml)$
   - repo: https://github.com/PyCQA/flake8
     rev: 7.0.0

Dockerfile

@@ -1,18 +1,19 @@
 FROM alpine
 
+ARG branch=main
 ENV GIT_REPO_URL=https://github.com/milinsoft/bank_app
 ENV PROJECT_FOLDER=/app/bank_app
 
 WORKDIR /app
 
 RUN set -ex \
     && apk add --update --no-cache git python3 py3-pip \
-    && git clone $GIT_REPO_URL $PROJECT_FOLDER -b refactor-workflow --depth=1 \
+    && git clone $GIT_REPO_URL $PROJECT_FOLDER -b $branch --depth=1 \
     && python3 -m venv venv \
-    && chmod +x ./venv/bin/activate \
-    && ./venv/bin/pip install -r $PROJECT_FOLDER/requirements.txt \
-    && ln -sf /venv/bin/python /usr/bin/python \
+    && source venv/bin/activate \
+    && pip install -r $PROJECT_FOLDER/requirements.txt \
     && apk del git \
     && rm -rf /var/cache/apk/* /root/.cache $PROJECT_FOLDER/.git
 
-CMD ["python", "-m", "unittest", "discover", "bank_app"]
+# Activate the virtual environment and run unit tests
+CMD ["sh", "-c", "source /app/venv/bin/activate && python -m unittest discover bank_app"]

__main__.py

@@ -1,9 +1,8 @@
 import sys
 from logging import getLogger
 
+from app.cli import BankAppCli
 from app.database import Database
-from app.interface import BankAppCli
-from settings import DB_URL
 
 _logger = getLogger(__name__)
 MAJOR = 3
@@ -18,10 +17,9 @@ def check_python_version(min_version=(MAJOR, MINOR)):
 
 if __name__ == "__main__":
     check_python_version()
-    db_session = Database(DB_URL).session
-    app = BankAppCli(db_session)
+    db = Database()
+    app = BankAppCli(db)
     try:
         app.main_menu()
     except KeyboardInterrupt:
-        db_session.close()
         print("\nGoodbye!")

app/cli/cli.py

@@ -0,0 +1,141 @@
+from datetime import date, datetime
+from logging import getLogger
+from os.path import exists
+
+from sqlalchemy.exc import SQLAlchemyError
+from tabulate import tabulate
+
+import settings
+from app.database import Database
+from app.domain_classes import AccountType
+from app.models import BankApp, Transaction
+from app.parsers import TransactionParser
+from app.services import AccountService, TransactionService
+from app.utils import UnitOfWork
+
+_logger = getLogger(__name__)
+
+
+class BankAppCli(BankApp):
+    def __init__(self, db: type[Database]) -> None:
+        self.acc_service = AccountService()
+        self.trx_service = TransactionService()
+        self.db = db
+        self.account_id: int | None = None
+        self.uow: UnitOfWork | None = UnitOfWork(db)  # imported instance
+        self.parser = TransactionParser()
+
+        self.menu_options = {
+            "0": ("Exit", self.exit_app),
+            "1": ("Import transactions (supported formats are: csv)", self.import_data),
+            "2": ("Show balance", self.show_balance),
+            "3": ("Search transactions for the a given period", self.search_transactions),
+        }
+        self.menu_msg = "\n".join(f"{k}: {v[0]}" for k, v in self.menu_options.items())
+
+    def main_menu(self) -> None:
+        self.pick_account()
+        while True:
+            choice = self.get_valid_action()
+            action = self.menu_options[choice][1]
+            action()
+
+    def import_data(self):
+        try:
+            trx_data = self.parser.parse_data(self.get_file_path())
+            _, balance = self.trx_service.create(self.uow, self.account_id, trx_data)
+            print(f"Transactions have been loaded successfully! Current balance: {balance}")
+        except (ValueError, SQLAlchemyError) as err:
+            _logger.error(err)
+
+    @classmethod
+    def get_file_path(cls) -> str:
+        while True:
+            file_path = input("Please provide the path to your file: ").strip("'\"")
+            if not exists(file_path):
+                print("Incorrect file path, please try again!")
+            else:
+                return file_path
+
+    def show_balance(self):
+        tar_get_date = self._get_date(mode="end_date")
+        print(
+            f"Your balance on {tar_get_date} is: ",
+            self.acc_service.get_balance(self.uow, self.account_id, tar_get_date),
+        )
+
+    def _search_transactions(self) -> list["Transaction"]:
+        return self.trx_service.get_by_date_range(
+            self.uow, self.account_id, self._get_date("start_date"), self._get_date("end_date")
+        )
+
+    def search_transactions(self) -> None:
+        transactions = self._search_transactions()
+        print(self._get_transaction_table(transactions) if transactions else "No transactions found!")
+
+    def get_valid_action(self):
+        print("\nPICK AN OPTION: ")
+        action = False
+        while action not in self.menu_options.keys():
+            action = input(f"{self.menu_msg}\n").strip()
+        return action
+
+    @staticmethod
+    def _get_date(mode: str):
+        if mode not in (allowed_modes := ("start_date", "end_date")):
+            raise ValueError(f"Invalid mode: {mode}. Allowed modes are {allowed_modes}")
+        today_date = date.today()
+
+        def compose__get_date_message() -> str:
+            date_example = datetime.strftime(today_date, settings.DATE_FORMAT)
+            action_description = (
+                "search from the oldest transaction\n"
+                if mode == allowed_modes[0]
+                else "pick today's date by default!\n"
+            )
+            return (
+                f"\nProvide the {mode} in the following {date_example} format or "
+                f"{action_description}\n press enter/return to {action_description}"
+            )
+
+        msg = compose__get_date_message()
+        while True:
+            tar_get_date = input(msg).strip()
+            if not tar_get_date:
+                return datetime.min.date() if mode == allowed_modes[0] else today_date
+            try:
+                tar_get_date = datetime.strptime(tar_get_date, settings.DATE_FORMAT).date()
+                if tar_get_date > today_date:
+                    tar_get_date = today_date
+            except ValueError:
+                _logger.error("Incorrect data format")
+            else:
+                return tar_get_date
+
+    @staticmethod
+    def _get_account_type() -> type[AccountType]:
+        acc_type: str | None | type[AccountType] = None
+        while not acc_type:
+            acc_type = getattr(
+                AccountType, input("Pick an account: Debit or Credit (debit/credit): ").upper().strip(), ""
+            )
+        return acc_type
+
+    def pick_account(self) -> None:
+        acc_type = self._get_account_type()
+        existing_account = self.acc_service.get_by_type(self.uow, acc_type)
+        self.account_id = existing_account.id if existing_account else self.acc_service.create_one(self.uow, acc_type)
+
+    @classmethod
+    def _get_transaction_table(cls, transactions: list[Transaction]) -> str:
+        """Return transactions in a tabular str format."""
+        return tabulate(
+            [(t.date, t.description, t.amount) for t in transactions],
+            headers=["Date", "Description", "Amount"],
+            colalign=("left", "left", "right"),
+            tablefmt="pretty",
+        )
+
+    @staticmethod
+    def exit_app():
+        exit(print("Goodbye!"))

app/database/database.py

@@ -3,9 +3,10 @@
 
 import settings
 from app.models import Base
+from app.utils import Singleton
 
 
-class Database:
+class Database(metaclass=Singleton):
     """DB connection abstraction.
 
     DB url format: ``dialect[+driver]://user:password@host/dbname[?key=value..]``,  # pragma: allowlist secret
@@ -19,9 +20,8 @@ class Database:
     def __init__(self, db_url: str = settings.DB_URL) -> None:
         self.db_url: str = db_url
         self.engine: Engine = create_engine(self.db_url)
-        self.session: Session = self.create_session()
-
-    def create_session(self) -> Session:
         # Create tables if they don't exist
         Base.metadata.create_all(self.engine)
+
+    def create_session(self) -> Session:
         return sessionmaker(bind=self.engine)()

app/domain_classes/__init__.py

@@ -0,0 +1,2 @@
+from .account_type import AccountType
+from .transaction_data import TransactionData

app/domain_classes/account_type.py

@@ -0,0 +1,6 @@
+from enum import Enum
+
+
+class AccountType(Enum):
+    CREDIT: int = 1
+    DEBIT: int = 2

app/domain_classes/transaction_data.py

@@ -0,0 +1,46 @@
+from dataclasses import dataclass
+from datetime import date, datetime
+from decimal import Decimal, InvalidOperation
+
+import settings
+
+
+@dataclass
+class TransactionData:
+    date: date
+    amount: Decimal
+    description: str
+    account_id: int | None = None
+
+    # noqa: D105
+    def __post_init__(self) -> None:
+        """Post init hooks."""
+        self._convert_str_to_date()
+        self._convert_str_to_decimal_amount()
+        self._check_description()
+
+    def set_account_id(self, account_id: int) -> type["TransactionData"]:
+        self.account_id = account_id
+        return self
+
+    def _check_description(self) -> None:
+        if not self.description:
+            raise ValueError("Missing transaction description!")
+
+    def _convert_str_to_date(self, date_format: str = settings.DATE_FORMAT) -> None:
+        try:
+            converted_date = datetime.strptime(self.date, date_format).date()
+        except ValueError:
+            raise ValueError(f"Wrong date format! Provided value: {self.date} Please use {date_format}")
+        if converted_date > date.today():
+            raise ValueError("Transaction date is in the future!")
+        self.date = converted_date
+
+    def _convert_str_to_decimal_amount(self, rounding: str = settings.ROUNDING) -> None:
+        try:
+            converted_amount = Decimal(self.amount).quantize(Decimal("0.00"), rounding=rounding)
+            if not converted_amount:
+                raise ValueError
+        except (ValueError, InvalidOperation):
+            raise ValueError("Incorrect transaction amount!")
+        self.amount = converted_amount