✨ PushLunch + LunchableApp

lunchable/_cli.py

@@ -43,7 +43,7 @@ class LunchMoneyContext(LunchableModel):
 )
 
 
-@click.group()
+@click.group(invoke_without_command=True)
 @click.version_option(
     version=lunchable.__version__, prog_name=lunchable.__application__
 )
@@ -57,6 +57,8 @@ def cli(ctx: click.core.Context, debug: bool, access_token: str) -> None:
     ctx.obj = LunchMoneyContext(debug=debug, access_token=access_token)
     traceback.install(show_locals=debug)
     set_up_logging(log_level=logging.DEBUG if debug is True else logging.INFO)
+    if ctx.invoked_subcommand is None:
+        click.echo(ctx.get_help())
 
 
 @cli.group()
@@ -387,8 +389,6 @@ def notify(continuous: bool, interval: int, user_key: str) -> None:
     push = PushLunch(user_key=user_key)
     if interval is not None:
         interval = int(interval)
-    if continuous is not None:
-        set_up_logging(log_level=logging.INFO)
     push.notify_uncleared_transactions(continuous=continuous, interval=interval)
 
 

lunchable/plugins/base/base_app.py

@@ -55,7 +55,7 @@ class LunchableDataContainer(BaseModel):
     user: UserObject = UserObject(
         user_id=0, user_name="", user_email="", account_id=0, budget_name=""
     )
-    crypto: Dict[int, UserObject] = {}
+    crypto: Dict[int, CryptoObject] = {}
 
     @property
     def asset_map(self) -> Dict[int, Union[PlaidAccountObject, AssetsObject]]:
@@ -71,6 +71,72 @@ def asset_map(self) -> Dict[int, Union[PlaidAccountObject, AssetsObject]]:
         asset_map.update(self.assets)
         return asset_map
 
+    @property
+    def plaid_accounts_list(self) -> List[PlaidAccountObject]:
+        """
+        List of Plaid Accounts
+
+        Returns
+        -------
+        List[PlaidAccountObject]
+        """
+        return list(self.plaid_accounts.values())
+
+    @property
+    def assets_list(self) -> List[AssetsObject]:
+        """
+        List of Assets
+
+        Returns
+        -------
+        List[AssetsObject]
+        """
+        return list(self.assets.values())
+
+    @property
+    def transactions_list(self) -> List[TransactionObject]:
+        """
+        List of Transactions
+
+        Returns
+        -------
+        List[TransactionObject]
+        """
+        return list(self.transactions.values())
+
+    @property
+    def categories_list(self) -> List[CategoriesObject]:
+        """
+        List of Categories
+
+        Returns
+        -------
+        List[CategoriesObject]
+        """
+        return list(self.categories.values())
+
+    @property
+    def tags_list(self) -> List[TagsObject]:
+        """
+        List of Tags
+
+        Returns
+        -------
+        List[TagsObject]
+        """
+        return list(self.tags.values())
+
+    @property
+    def crypto_list(self) -> List[CryptoObject]:
+        """
+        List of Crypto
+
+        Returns
+        -------
+        List[CryptoObject]
+        """
+        return list(self.crypto.values())
+
 
 class BaseLunchableApp(ABC):
     """
@@ -234,17 +300,70 @@ def get_latest_cache(
             )
 
     def refresh_transactions(
-        self, start_date: datetime.date, end_date: datetime.date
+        self,
+        start_date: Optional[Union[datetime.date, datetime.datetime, str]] = None,
+        end_date: Optional[Union[datetime.date, datetime.datetime, str]] = None,
+        tag_id: Optional[int] = None,
+        recurring_id: Optional[int] = None,
+        plaid_account_id: Optional[int] = None,
+        category_id: Optional[int] = None,
+        asset_id: Optional[int] = None,
+        group_id: Optional[int] = None,
+        is_group: Optional[bool] = None,
+        status: Optional[str] = None,
+        offset: Optional[int] = None,
+        limit: Optional[int] = None,
+        debit_as_negative: Optional[bool] = None,
+        pending: Optional[bool] = None,
+        params: Optional[Dict[str, Any]] = None,
     ) -> Dict[int, TransactionObject]:
         """
         Refresh App data with the latest transactions
 
+        start_date: Optional[Union[datetime.date, datetime.datetime, str]]
+            Denotes the beginning of the time period to fetch transactions for. Defaults
+            to beginning of current month. Required if end_date exists. Format: YYYY-MM-DD.
+        end_date: Optional[Union[datetime.date, datetime.datetime, str]]
+            Denotes the end of the time period you'd like to get transactions for.
+            Defaults to end of current month. Required if start_date exists.
+        tag_id: Optional[int]
+            Filter by tag. Only accepts IDs, not names.
+        recurring_id: Optional[int]
+            Filter by recurring expense
+        plaid_account_id: Optional[int]
+            Filter by Plaid account
+        category_id: Optional[int]
+            Filter by category. Will also match category groups.
+        asset_id: Optional[int]
+            Filter by asset
+        group_id: Optional[int]
+            Filter by group_id (if the transaction is part of a specific group)
+        is_group: Optional[bool]
+            Filter by group (returns transaction groups)
+        status: Optional[str]
+            Filter by status (Can be cleared or uncleared. For recurring
+            transactions, use recurring)
+        offset: Optional[int]
+            Sets the offset for the records returned
+        limit: Optional[int]
+            Sets the maximum number of records to return. Note: The server will not
+            respond with any indication that there are more records to be returned.
+            Please check the response length to determine if you should make another
+            call with an offset to fetch more transactions.
+        debit_as_negative: Optional[bool]
+            Pass in true if you'd like expenses to be returned as negative amounts and
+            credits as positive amounts. Defaults to false.
+        pending: Optional[bool]
+            Pass in true if you'd like to include imported transactions with a pending status.
+        params: Optional[dict]
+            Additional Query String Params
+
         Returns
         -------
         Dict[int, TransactionObject]
         """
         transactions = self.lunch.get_transactions(
-            start_date=start_date, end_date=end_date
+            start_date=start_date, end_date=end_date, status=status
         )
         transaction_map = {item.id: item for item in transactions}
         self.lunch_data.transactions = transaction_map

lunchable/plugins/pushlunch/pushover.py

@@ -2,7 +2,6 @@
 Pushover Notifications via lunchable
 """
 
-import datetime
 import logging
 from base64 import b64decode
 from json import loads
@@ -13,8 +12,13 @@
 
 import requests
 
-from lunchable import LunchMoney
-from lunchable.models import AssetsObject, PlaidAccountObject, TransactionObject
+from lunchable.models import (
+    AssetsObject,
+    CategoriesObject,
+    PlaidAccountObject,
+    TransactionObject,
+)
+from lunchable.plugins import LunchableApp
 
 logger = logging.getLogger(__name__)
 
@@ -27,7 +31,7 @@ class PushLunchError(Exception):
     pass
 
 
-class PushLunch:
+class PushLunch(LunchableApp):
     """
     Lunch Money Pushover Notifications via Lunchable
     """
@@ -39,7 +43,6 @@ def __init__(
         user_key: Optional[str] = None,
         app_token: Optional[str] = None,
         lunchmoney_access_token: Optional[str] = None,
-        lunchable_client: Optional[LunchMoney] = None,
     ):
         """
         Initialize
@@ -55,11 +58,10 @@ def __init__(
         lunchmoney_access_token: Optional[str]
             LunchMoney Access Token. Will be inherited from `LUNCHMONEY_ACCESS_TOKEN`
             environment variable.
-        lunchable_client: Optional[LunchMoney]
-            lunchable client to use. One will be created if none provided.
         """
-        self.session = requests.Session()
-        self.session.headers.update({"Content-Type": "application/json"})
+        super().__init__(access_token=lunchmoney_access_token)
+        self.pushover_session = requests.Session()
+        self.pushover_session.headers.update({"Content-Type": "application/json"})
 
         _courtesy_token = b"YXpwMzZ6MjExcWV5OGFvOXNicWF0cmdraXc4aGVz"
         if app_token is None:
@@ -72,11 +74,9 @@ def __init__(
                 "a `PUSHOVER_USER_KEY` environment variable"
             )
         self._params = {"user": user_key, "token": token}
-        self.lunchable = lunchable_client or LunchMoney(
-            access_token=lunchmoney_access_token
+        self.get_latest_cache(
+            include=[AssetsObject, PlaidAccountObject, CategoriesObject]
         )
-        self.asset_mapping = self._get_assets()
-        self.category_mapping = self._get_categories()
         self.notified_transactions: List[int] = []
 
     def send_notification(
@@ -145,7 +145,7 @@ def send_notification(
             key: value for key, value in params_dict.items() if value is not None
         }
         params.update(self._params)
-        response = self.session.post(url=self.pushover_endpoint, params=params)
+        response = self.pushover_session.post(url=self.pushover_endpoint, params=params)
         response.raise_for_status()
         return response
 
@@ -171,16 +171,21 @@ class hasn't already posted this particular notification
         if transaction.category_id is None:
             category = "N/A"
         else:
-            category = self.category_mapping[transaction.category_id]
+            category = self.lunch_data.categories[transaction.category_id].name
         account_id = transaction.plaid_account_id or transaction.asset_id
         assert account_id is not None
+        account = self.lunch_data.asset_map[account_id]
+        if isinstance(account, AssetsObject):
+            account_name = account.display_name or account.name
+        else:
+            account_name = account.name
         transaction_formatted = dedent(
             f"""
         <b>Payee:</b> <i>{transaction.payee}</i>
         <b>Amount:</b> <i>{self._format_float(transaction.amount)}</i>
         <b>Date:</b> <i>{transaction.date.strftime("%A %B %-d, %Y")}</i>
         <b>Category:</b> <i>{category}</i>
-        <b>Account:</b> <i>{self.asset_mapping[account_id]}</i>
+        <b>Account:</b> <i>{account_name}</i>
         """
         ).strip()
         if transaction.currency is not None:
@@ -208,43 +213,6 @@ class hasn't already posted this particular notification
         self.notified_transactions.append(transaction.id)
         return loads(response.content)
 
-    def _get_assets(self) -> Dict[int, str]:
-        """
-        Get Mapping Of Asset ID -> Asset Name
-
-        Returns
-        -------
-        Dict[int, str]
-        """
-        manual_assets = self.lunchable.get_assets()
-        plaid_account = self.lunchable.get_plaid_accounts()
-        assets = [*manual_assets, *plaid_account]
-        asset_mapping = {}
-        for account in assets:
-            if isinstance(account, AssetsObject):
-                if account.display_name is None:
-                    name = account.name
-                else:
-                    name = account.display_name
-                asset_mapping[account.id] = name
-            elif isinstance(account, PlaidAccountObject):
-                asset_mapping[account.id] = account.name
-        return asset_mapping
-
-    def _get_categories(self) -> Dict[int, str]:
-        """
-        Get Mapping Of Category ID -> Category Name
-
-        Returns
-        -------
-        Dict[int, str]
-        """
-        categories = self.lunchable.get_categories()
-        category_mapping = {}
-        for category in categories:
-            category_mapping[category.id] = category.name
-        return category_mapping
-
     @classmethod
     def _format_float(cls, amount: float) -> str:
         """
@@ -265,23 +233,6 @@ def _format_float(cls, amount: float) -> str:
             float_string = "$ {:,.2f}".format(float(amount))
         return float_string
 
-    def _get_uncleared_transactions(
-        self,
-        start_date: Optional[datetime.datetime] = None,
-        end_date: Optional[datetime.datetime] = None,
-    ) -> List[TransactionObject]:
-        """
-        Get Uncleared Transactions
-
-        Returns
-        -------
-        List[TransactionObject]
-        """
-        uncleared_transactions = self.lunchable.get_transactions(
-            start_date=start_date, end_date=end_date, status="uncleared"
-        )
-        return uncleared_transactions
-
     def notify_uncleared_transactions(
         self, continuous: bool = False, interval: Optional[int] = None
     ) -> List[TransactionObject]:
@@ -316,7 +267,7 @@ def notify_uncleared_transactions(
 
         while continuous_search is True:
             found_transactions = len(self.notified_transactions)
-            uncleared_transactions += self._get_uncleared_transactions()
+            uncleared_transactions += self.lunch.get_transactions(status="uncleared")
             for transaction in uncleared_transactions:
                 self.post_transaction(transaction=transaction)
             if continuous is True: