[Fix] (Breaking Change) better typing

Makefile

@@ -1,16 +1,13 @@
 
 VENV?=${VIRTUAL_ENV}
-PYTHONPATH=./src
-
 
 ${VENV}/bin/activate:
 	python3.10 -m venv ${VENV}
 
 venv: ${VENV}/bin/activate
 
 install: requirements.txt requirements-dev.txt venv
-	${VENV}/bin/pip3 install -r requirements.txt
-	${VENV}/bin/pip3 install -r requirements-dev.txt
+	${VENV}/bin/pip3 install -e .
 
 test:
 	${VENV}/bin/python -m pytest tests

README.md

@@ -3,12 +3,10 @@
 ![Static Badge](https://img.shields.io/badge/Python-3-blue?style=flat&logo=Python)
 ![PyPI](https://img.shields.io/pypi/v/moneywiz-api)
 
-
 <a href="https://www.buymeacoffee.com/Ileodo" target="_blank"><img src="https://cdn.buymeacoffee.com/buttons/v2/default-blue.png" alt="Buy Me A Coffee" style="height: 60px !important;width: 217px !important;" ></a>
 
 A Python API to access MoneyWiz Sqlite database.
 
-
 ## Get Started
 
 ```bash
@@ -42,6 +40,8 @@ print(record)
 
 ```
 
+It also offers a interactive shell `moneywiz-cli`.
+
 ## Contribution
 
 This project is in very early stage, all contributions are welcomed!

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "moneywiz-api"
-version = "0.2.1"
+version = "1.0.0"
 authors = [
     { name="iLeoDo", email="iLeoDo@gmail.com" },
 ]

src/moneywiz_api/cli/cli.py

@@ -1,10 +1,13 @@
 from os.path import expanduser
 from pathlib import Path
-from code import interact
+from code import InteractiveConsole
 from typing import Dict, List
 
 import click
 import logging
+import readline
+import rlcompleter
+import random
 
 from moneywiz_api.cli.helpers import ShellHelper
 from moneywiz_api.moneywiz_api import MoneywizApi
@@ -24,6 +27,12 @@ def get_default_path() -> Path:
     type=click.Path(writable=False, readable=True, exists=True),
     default=get_default_path(),
 )
+@click.option(
+    "-d",
+    "--demo-dump",
+    is_flag=True,
+    help="Dumy some demo data",
+)
 @click.option(
     "--log-level",
     default="INFO",
@@ -32,7 +41,7 @@ def get_default_path() -> Path:
     ),
     help="Set the logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)",
 )
-def main(db_file_path, log_level):
+def main(db_file_path, demo_dump, log_level):
     """
     Interactive shell to access MoneyWiz (Read-only)
     """
@@ -86,7 +95,37 @@ def main(db_file_path, log_level):
         "===================================================================",
     )
 
-    interact(local=locals(), banner="\n".join(banner))
+    if demo_dump:
+        _users_table = helper.users_table()
+        click.secho("Users Table", fg="yellow")
+        click.secho("--------------------------------", fg="yellow")
+        click.secho(_users_table.to_string(index=False))
+        click.secho("--------------------------------\n", fg="yellow")
+
+        _userid_list = _users_table["id"].tolist()
+        _userid_list.remove(1)
+        _user_id = random.choice(_userid_list)
+
+        _categories_table = helper.categories_table(_user_id)
+        click.secho(f"Categories Table for User {_user_id}", fg="yellow")
+        click.secho("--------------------------------", fg="yellow")
+        click.secho(
+            _categories_table[["id", "name", "type"]].sample(5).to_string(index=False)
+        )
+        click.secho("--------------------------------\n", fg="yellow")
+
+        _accounts_table = helper.accounts_table(_user_id)
+        click.secho(f"Accounts Table for User {_user_id}", fg="yellow")
+        click.secho("--------------------------------", fg="yellow")
+        click.secho(_accounts_table[["id", "name"]].sample(5).to_string(index=False))
+        click.secho("--------------------------------\n", fg="yellow")
+
+    _vars = globals()
+    _vars.update(locals())
+
+    readline.set_completer(rlcompleter.Completer(_vars).complete)
+    readline.parse_and_bind("tab: complete")
+    InteractiveConsole(_vars).interact(banner="\n".join(banner))
 
 
 if __name__ == "__main__":

src/moneywiz_api/cli/helpers.py

@@ -1,14 +1,15 @@
 from dataclasses import asdict
 from typing import Dict
+from pathlib import Path
 
 import logging
-import pathlib
 import json
 import click
 import pandas as pd
 
 from moneywiz_api import MoneywizApi
-from moneywiz_api.types import ID
+from moneywiz_api.managers.record_manager import RecordManager
+from moneywiz_api.types import ID, GID
 
 
 logger = logging.getLogger(__name__)
@@ -24,18 +25,13 @@ def view_id(self, record_id: ID):
         click.echo(self._mw_api.accessor.typename_for(record.ent()))
         click.echo(json.dumps(record.filtered(), sort_keys=True, indent=4))
 
-    def print_investment_holdings_for_account(self, account_id: ID):
-        for x in [
-            f"{h.id}: 0, #{h.symbol}"
-            for h in self._mw_api.investment_holding_manager.get_holdings_for_account(
-                account_id
-            )
-        ]:
-            print(x)
-
-    def write_state_data_files(self):
-        path_prefix = "data/state"
-        pathlib.Path(path_prefix).mkdir(parents=True, exist_ok=True)
+    def view_gid(self, record_gid: GID):
+        record = self._mw_api.accessor.get_record_by_gid(record_gid)
+        click.echo(self._mw_api.accessor.typename_for(record.ent()))
+        click.echo(json.dumps(record.filtered(), sort_keys=True, indent=4))
+
+    def write_stats_data_files(self, path_prefix: Path = Path("data/stats")):
+        Path(path_prefix).mkdir(parents=True, exist_ok=True)
         managers_map: Dict[str, object] = {
             "ent": self._mw_api.accessor,
             "account": self._mw_api.account_manager,
@@ -47,38 +43,46 @@ def write_state_data_files(self):
 
         for name, obj in managers_map.items():
             with open(f"{path_prefix}/{name}.data", "w", encoding="utf-8") as file:
-                print(obj, file=file)
+                click.echo(obj, file=file)
 
-    def account_tables(self, user_id: ID):
-        pd.set_option("display.max_colwidth", None)
-        pd.set_option("display.max_rows", None)
+    def pd_table(self, manager: RecordManager) -> pd.DataFrame:
+        records = manager.records().values()
+        df = pd.DataFrame.from_records([r.as_dict() for r in records])
+        return df
 
-        accounts = self._mw_api.account_manager.get_accounts_for_user(user_id)
+    def users_table(self) -> pd.DataFrame:
+        users = self._mw_api.accessor.get_users()
+        records = [
+            {"id": id, "login_name": login_name} for id, login_name in users.items()
+        ]
+        return pd.DataFrame.from_records(records).sort_values(by=["id"], ascending=True)
 
-        df = pd.DataFrame.from_records([asdict(account) for account in accounts])
-        print(
-            df.sort_values(by=["_group_id", "_display_order"])[
-                ["id", "name"]
-            ].to_string(index=False)
-        )
+    def categories_table(self, user_id: ID) -> pd.DataFrame:
+        categories = self._mw_api.category_manager.get_categories_for_user(user_id)
+        return pd.DataFrame.from_records(
+            [category.as_dict() for category in categories]
+        ).sort_values(by=["id"], ascending=True)
 
-    def transactions_for_account(self, account_id: ID):
-        records = self._mw_api.transaction_manager.get_all_for_account(account_id)
+    def accounts_table(self, user_id: ID) -> pd.DataFrame:
+        # pd.set_option("display.max_colwidth", None)
+        # pd.set_option("display.max_rows", None)
 
-        df = pd.DataFrame.from_records([asdict(record) for record in records])
-        print(
-            df.sort_values(by=["date"], ascending=False)[
-                ["id", "date", "amount"]
-            ].to_string(index=False)
-        )
+        accounts = self._mw_api.account_manager.get_accounts_for_user(user_id)
+        return pd.DataFrame.from_records(
+            [account.as_dict() for account in accounts]
+        ).sort_values(by=["group_id", "display_order"])
 
-        print("total_amount:", df["amount"].sum())
-        print("number of transactions:", df.shape[0])
+    def investment_holdings_table(self, account_id: ID) -> pd.DataFrame:
+        investment_holdings = (
+            self._mw_api.investment_holding_manager.get_holdings_for_account(account_id)
+        )
+        return pd.DataFrame.from_records(
+            [investment_holding.as_dict() for investment_holding in investment_holdings]
+        ).sort_values(by=["account", "symbol"])
 
-    def generate_categories_list(self, user_id: ID):
-        categories = self._mw_api.category_manager.get_categories_for_user(user_id)
+    def transactions_table(self, account_id: ID) -> pd.DataFrame:
+        records = self._mw_api.transaction_manager.get_all_for_account(account_id)
 
-        for category in categories:
-            print(
-                f"""{category.id}: ["{category.type}", {", ".join(['"'+ x.replace(" ","") + '"' for x in category_manager.get_name_chain(category.id)])}],"""
-            )
+        return pd.DataFrame.from_records(
+            [record.as_dict() for record in records]
+        ).sort_values(by=["datetime"], ascending=False)

src/moneywiz_api/database_accessor.py

@@ -2,9 +2,11 @@
 from collections import defaultdict
 from pathlib import Path
 from typing import Dict, List, Any, Callable, Tuple
+from decimal import Decimal
 
 from moneywiz_api.model.record import Record
-from moneywiz_api.types import ENT_ID, ID
+from moneywiz_api.model.raw_data_handler import RawDataHandler as RDH
+from moneywiz_api.types import ENT_ID, ID, GID
 
 
 class DatabaseAccessor:
@@ -70,8 +72,20 @@ def get_record(self, pk_id: ID, constructor: Callable = Record):
 
         return constructor(res.fetchone())
 
-    def get_category_assignment(self) -> Dict[ID, List[Tuple[ID, float]]]:
-        transaction_map: Dict[ID, List[Tuple[ID, float]]] = defaultdict(list)
+    def get_record_by_gid(self, gid: GID, constructor: Callable = Record):
+        cur = self._con.cursor()
+        res = cur.execute(
+            """
+        SELECT * FROM ZSYNCOBJECT WHERE ZGID = ?
+        
+        """,
+            [gid],
+        )
+
+        return constructor(res.fetchone())
+
+    def get_category_assignment(self) -> Dict[ID, List[Tuple[ID, Decimal]]]:
+        transaction_map: Dict[ID, List[Tuple[ID, Decimal]]] = defaultdict(list)
         cur = self._con.cursor()
         res = cur.execute(
             """
@@ -81,7 +95,7 @@ def get_category_assignment(self) -> Dict[ID, List[Tuple[ID, float]]]:
         )
         for row in res.fetchall():
             transaction_map[row["ZTRANSACTION"]].append(
-                (row["ZCATEGORY"], row["ZAMOUNT"])
+                (row["ZCATEGORY"], RDH.get_decimal(row, "ZAMOUNT"))
             )
         return transaction_map
 
@@ -110,3 +124,16 @@ def get_tags_map(self) -> Dict[ID, List[ID]]:
         for row in res.fetchall():
             transactions_to_tags[row["Z_36TRANSACTIONS"]].append(row["Z_35TAGS"])
         return transactions_to_tags
+
+    def get_users(self) -> Dict[ID, str]:
+        users_map: Dict[ID, str] = {}
+        cur = self._con.cursor()
+        res = cur.execute(
+            """
+        SELECT Z_PK, ZSYNCLOGIN FROM  "ZUSER"
+        
+        """
+        )
+        for row in res.fetchall():
+            users_map[row["Z_PK"]] = row["ZSYNCLOGIN"]
+        return users_map

src/moneywiz_api/managers/account_manager.py

@@ -32,12 +32,10 @@ def ents(self) -> Dict[str, Callable]:
         }
 
     def records(self) -> Dict[ID, Account]:
-        return dict(
-            sorted(super().records().items(), key=lambda x: x[1]._display_order)
-        )
+        return dict(sorted(super().records().items(), key=lambda x: x[1].display_order))
 
     def get_accounts_for_user(self, user_id: ID) -> List[Account]:
         return sorted(
             [x for _, x in self.records().items() if x.user == user_id],
-            key=lambda x: (x._group_id, x._display_order),
+            key=lambda x: (x.group_id, x.display_order),
         )

src/moneywiz_api/managers/category_manager.py

@@ -20,10 +20,10 @@ def get_name_chain(self, category_id: ID) -> List[str]:
         current = self.get(category_id)
         while current:
             ret.insert(0, current.name)
-            if not current.parentId:
+            if not current.parent_id:
                 break
             else:
-                current = self.get(current.parentId)
+                current = self.get(current.parent_id)
         return ret
 
     def get_name_chain_by_gid(self, category_gid: GID) -> List[str]:

src/moneywiz_api/managers/investment_holding_manager.py

@@ -1,4 +1,5 @@
 from typing import Dict, Callable, List
+from decimal import Decimal
 
 from moneywiz_api.model.investment_holding import InvestmentHolding
 from moneywiz_api.managers.record_manager import RecordManager
@@ -18,8 +19,8 @@ def ents(self) -> Dict[str, Callable]:
     def get_holdings_for_account(self, account_id: ID) -> List[InvestmentHolding]:
         return [x for _, x in self.records().items() if x.account == account_id]
 
-    def update_last_price(self, latest_price: float):
+    def update_last_price(self, latest_price: Decimal):
         raise NotImplementedError()
 
-    def update_price_table(self, latest_price: float):
+    def update_price_table(self, latest_price: Decimal):
         raise NotImplementedError()

src/moneywiz_api/managers/record_manager.py

@@ -31,7 +31,7 @@ def load(self, db_accessor: DatabaseAccessor) -> None:
     def add(self, record: T) -> None:
         self._records[record.id] = record
         if record.gid in self._gid_to_id:
-            raise Exception(
+            raise RuntimeError(
                 f"Duplicate gid for {record}, existing record Id {self._gid_to_id[record.gid]}"
             )