commit b49271fd252daca2424701499990e422886d8e13
parent 0abe35d8af9ee6a73fa49a2cbeba6f522fcf58bb
Author: Carlosokumu <carlosokumu254@gmail.com>
Date: Thu, 19 Feb 2026 19:39:44 +0300
initialize core submodule
Diffstat:
3 files changed, 151 insertions(+), 0 deletions(-)
diff --git a/dummy/usawa/core/__init__.py b/dummy/usawa/core/__init__.py
diff --git a/dummy/usawa/core/entry_service.py b/dummy/usawa/core/entry_service.py
@@ -0,0 +1,60 @@
+import logging
+from datetime import datetime
+from typing import Optional
+import uuid
+
+from usawa.service import UnixClient
+from usawa.storage.entry_mapper import EntryMapper
+from .models import LedgerEntry
+from usawa.storage.ledger_repository import LedgerRepository
+
+logg = logging.getLogger("core.entry_service")
+
+
+class EntryService:
+ """Business logic for ledger entries"""
+
+ def __init__(self, repository: LedgerRepository,unixClient: UnixClient):
+ self.repository = repository
+ self.unix_client = unixClient
+
+ def save_entry(self, entry: LedgerEntry) -> bool:
+ """
+ Save entry with business logic
+
+ :param entry: Entry to save
+ :type entry: LedgerEntry
+ :return: True if saved successfully
+ :rtype: bool
+ """
+
+ entry.tx_date = datetime.now()
+ entry.date_registered = datetime.now()
+ entry.parent_digest = self._get_parent_digest()
+ entry.unit_index = self._get_unit_index()
+ entry.transaction_ref = self._generate_transaction_ref()
+ entry.serial = self.repository.get_next_serial()
+
+ is_valid, error_msg = entry.validate()
+ if not is_valid:
+ logg.error(f"Entry validation failed: {error_msg}")
+ return False
+
+ return self.repository.save(entry)
+
+
+ def _get_next_serial(self) -> int:
+ """Get next serial number"""
+ return self.repository.get_max_serial() + 1
+
+ def _get_parent_digest(self) -> str:
+ """Get digest of previous ledger state"""
+ return self.repository._get_parent_digest()
+
+ def _get_unit_index(self) -> int:
+ """Get Unix timestamp for unit validation"""
+ return int(datetime.now().timestamp())
+
+ def _generate_transaction_ref(self) -> str:
+ """Generate UUID for transaction"""
+ return str(uuid.uuid4())
+\ No newline at end of file
diff --git a/dummy/usawa/core/models.py b/dummy/usawa/core/models.py
@@ -0,0 +1,90 @@
+from dataclasses import dataclass,field
+from typing import Optional,List, Union
+from datetime import datetime
+from pathlib import Path
+
+@dataclass
+class LedgerEntry:
+ """Ledger entry data model"""
+
+ # Basic details
+ external_reference: Optional[str] = None
+ description: Optional[str] = None
+
+ # Transaction details
+ amount: float = 0.0
+ source_unit: str = ""
+ source_type: str = ""
+ source_path: str = "general"
+ dest_unit: str = ""
+ dest_type: str = ""
+ dest_path: str = "general"
+
+ # Attachmentments
+ attachments: List[str] = field(default_factory=list)
+
+ # Metadata (auto-generated)
+ serial: Optional[int] = None
+ tx_date: Optional[datetime] = None
+ date_registered: Optional[datetime] = None
+ parent_digest: Optional[str] = None
+ unit_index: Optional[int] = None
+
+ def validate(self) -> tuple[bool, str]:
+ """Validate entry data"""
+ if self.amount <= 0:
+ return False, "Amount must be greater than 0"
+
+ if not self.source_unit or not self.dest_unit:
+ return False, "Unit/Currency is required for both source and destination"
+
+ if not self.source_type or not self.dest_type:
+ return False, "Account type is required for both source and destination"
+
+ if not self.source_path or not self.dest_path:
+ return False, "Account path is required for both source and destination"
+
+ # Validate attachments
+ if self.attachments:
+ for filepath in self.attachments:
+ if not Path(filepath).exists():
+ return False, f"Attachment file not found: {filepath}"
+
+ return True, ""
+
+ def __repr__(self):
+ return (
+ f"LedgerEntry("
+ f"external_reference={self.external_reference!r}, "
+ f"description={self.description!r}, "
+ f"serial={self.serial!r}, "
+ f"amount={self.amount}, "
+ f"source_unit={self.source_unit}, "
+ f"source_type={self.source_type}, "
+ f"dest_unit={self.dest_unit}, "
+ f"dest_type={self.dest_type})"
+ f"attachments={self.attachments!r})"
+ )
+
+ def add_attachment(self, filepath: Union[str, List[str]]):
+ """
+ Add one or more attachment file paths
+ """
+ if isinstance(filepath, str):
+ if filepath not in self.attachments:
+ self.attachments.append(filepath)
+ elif isinstance(filepath, list):
+ for path in filepath:
+ if path not in self.attachments:
+ self.attachments.append(path)
+ else:
+ raise TypeError(f"filepath must be str or List[str], got {type(filepath)}")
+
+ def remove_attachment(self, filepath: str):
+ """Remove an attachment file path"""
+ if filepath in self.attachments:
+ self.attachments.remove(filepath)
+
+ def get_attachment_count(self) -> int:
+ """Get number of attachments"""
+ return len(self.attachments)
