Use new Cora Credit Cart importer

Author: rennerocha

thebook/bookkeeping/importers/__init__.py

@@ -4,6 +4,8 @@
 
 from thebook.bookkeeping.importers.csv import CSVCoraCreditCardImporter, CSVImporter
 from thebook.bookkeeping.importers.ofx import OFXImporter
+from thebook.bookkeeping.models import Transaction
+from thebook.integrations.cora.credit_card_invoice import CoraCreditCardInvoiceImporter
 
 logger = logging.getLogger(__name__)
 
@@ -23,7 +25,7 @@ def import_transactions(
 ):
     importers = {
         "csv": CSVImporter,
-        "csv_cora_credit_card": CSVCoraCreditCardImporter,
+        "csv_cora_credit_card": CoraCreditCardInvoiceImporter,
         "ofx": OFXImporter,
     }
     importer = importers.get(file_type) or None
@@ -33,7 +35,19 @@ def import_transactions(
         )
 
     try:
-        importer(transactions_file, bank_account, user).run(start_date, end_date)
+        if file_type == "csv_cora_credit_card":
+            csv_importer = importer(transactions_file)
+            transactions = csv_importer.get_transactions(
+                start_date, end_date, exclude_existing=True
+            )
+            Transaction.objects.bulk_create(
+                transactions,
+                update_conflicts=True,
+                update_fields=["description", "amount"],
+                unique_fields=["reference"],
+            )
+        else:
+            importer(transactions_file, bank_account, user).run(start_date, end_date)
     except Exception as err:
         logger.exception(err)
         raise ImportTransactionsError(_("Something wrong happened during file import."))

thebook/integrations/cora/credit_card_invoice.py

@@ -59,7 +59,7 @@ def get_transactions(
     ) -> list[Transaction]:
         transactions = []
 
-        csv_content = self.invoice_file.read()
+        csv_content = self.invoice_file.read().decode()
         reader = csv.DictReader(io.StringIO(csv_content))
 
         if set(reader.fieldnames) != self._expected_field_names:
@@ -82,7 +82,10 @@ def get_transactions(
             if (
                 exclude_existing
                 and Transaction.objects.filter(
-                    date=transaction_date, description=description, amount=amount
+                    date=transaction_date,
+                    description=description,
+                    amount=amount,
+                    bank_account=self.bank_account,
                 ).exists()
             ):
                 continue

thebook/integrations/tests/cora/test_credit_card_invoice.py

@@ -42,7 +42,7 @@ def test_valid_but_empty_invoice_file(
         invoice_file_path = (
             request.path.parent / "data" / "empty-cora-credit-card-invoice.csv"
         )
-        with open(invoice_file_path, "r") as invoice_file:
+        with open(invoice_file_path, "rb") as invoice_file:
             importer = CoraCreditCardInvoiceImporter(invoice_file)
             transactions = importer.get_transactions()
             assert transactions == []
@@ -51,7 +51,7 @@ def test_invalid_invoice_file(self, db, request):
         invoice_file_path = (
             request.path.parent / "data" / "invalid-cora-credit-card-invoice.csv"
         )
-        with open(invoice_file_path, "r") as invoice_file:
+        with open(invoice_file_path, "rb") as invoice_file:
             importer = CoraCreditCardInvoiceImporter(invoice_file)
 
             with pytest.raises(InvalidCoraCreditCardInvoice):
@@ -65,7 +65,7 @@ def test_one_transaction_excluding_existing_false(
             / "data"
             / "cora-credit-card-invoice-one-transaction.csv"
         )
-        with open(invoice_file_path, "r") as invoice_file:
+        with open(invoice_file_path, "rb") as invoice_file:
             importer = CoraCreditCardInvoiceImporter(invoice_file)
             transactions = importer.get_transactions()
 
@@ -81,58 +81,87 @@ def test_one_transaction_excluding_existing_false(
             assert transactions[0].source == "cora-credit-card-invoice-importer"
             assert transactions[0].created_by == user
 
-    def test_one_transaction_excluding_existing_true(self, db, request):
+    def test_one_transaction_excluding_existing_true(
+        self, db, request, cora_credit_card_bank_account
+    ):
         baker.make(
             Transaction,
             date=datetime.date(2026, 2, 21),
             description="BACKBLAZE INC",
             amount=decimal.Decimal("-8.91"),
+            bank_account=cora_credit_card_bank_account,
         )
 
         invoice_file_path = (
             request.path.parent
             / "data"
             / "cora-credit-card-invoice-one-transaction.csv"
         )
-        with open(invoice_file_path, "r") as invoice_file:
+        with open(invoice_file_path, "rb") as invoice_file:
             importer = CoraCreditCardInvoiceImporter(invoice_file)
             transactions = importer.get_transactions(exclude_existing=True)
             assert len(transactions) == 0
 
-    def test_two_transactions_excluding_existing_true(self, db, request):
+    def test_one_transaction_from_different_bank_account(
+        self, db, request, cora_credit_card_bank_account
+    ):
+        baker.make(
+            Transaction,
+            date=datetime.date(2026, 2, 21),
+            description="BACKBLAZE INC",
+            amount=decimal.Decimal("-8.91"),
+        )
+
+        invoice_file_path = (
+            request.path.parent
+            / "data"
+            / "cora-credit-card-invoice-one-transaction.csv"
+        )
+        with open(invoice_file_path, "rb") as invoice_file:
+            importer = CoraCreditCardInvoiceImporter(invoice_file)
+            transactions = importer.get_transactions(exclude_existing=True)
+            assert len(transactions) == 1
+
+    def test_two_transactions_excluding_existing_true(
+        self, db, request, cora_credit_card_bank_account
+    ):
         baker.make(
             Transaction,
             date=datetime.date(2026, 2, 21),
             description="BACKBLAZE INC",
             amount=decimal.Decimal("-8.91"),
+            bank_account=cora_credit_card_bank_account,
         )
 
         invoice_file_path = (
             request.path.parent
             / "data"
             / "cora-credit-card-invoice-two-transactions.csv"
         )
-        with open(invoice_file_path, "r") as invoice_file:
+        with open(invoice_file_path, "rb") as invoice_file:
             importer = CoraCreditCardInvoiceImporter(invoice_file)
             transactions = importer.get_transactions(exclude_existing=True)
 
             assert len(transactions) == 1
             assert transactions[0].description == "VULTR BY CONSTANT"
 
-    def test_two_transactions_excluding_existing_false(self, db, request):
+    def test_two_transactions_excluding_existing_false(
+        self, db, request, cora_credit_card_bank_account
+    ):
         baker.make(
             Transaction,
             date=datetime.date(2026, 2, 21),
             description="BACKBLAZE INC",
             amount=decimal.Decimal("-8.91"),
+            bank_account=cora_credit_card_bank_account,
         )
 
         invoice_file_path = (
             request.path.parent
             / "data"
             / "cora-credit-card-invoice-two-transactions.csv"
         )
-        with open(invoice_file_path, "r") as invoice_file:
+        with open(invoice_file_path, "rb") as invoice_file:
             importer = CoraCreditCardInvoiceImporter(invoice_file)
             transactions = importer.get_transactions(exclude_existing=False)
 
@@ -173,7 +202,7 @@ def test_transactions_filtered_by_date_range(
             / "data"
             / "cora-credit-card-invoice-multiple-transactions.csv"
         )
-        with open(invoice_file_path, "r") as invoice_file:
+        with open(invoice_file_path, "rb") as invoice_file:
             importer = CoraCreditCardInvoiceImporter(invoice_file)
             transactions = importer.get_transactions(
                 start_date=start_date, end_date=end_date