diff --git a/lunchable/_cli.py b/lunchable/_cli.py
index 6401a98..1efba11 100644
--- a/lunchable/_cli.py
+++ b/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)
diff --git a/lunchable/plugins/base/base_app.py b/lunchable/plugins/base/base_app.py
index fbb9344..35cca16 100644
--- a/lunchable/plugins/base/base_app.py
+++ b/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
diff --git a/lunchable/plugins/pushlunch/pushover.py b/lunchable/plugins/pushlunch/pushover.py
index 1cfc745..ff2c9f7 100644
--- a/lunchable/plugins/pushlunch/pushover.py
+++ b/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"""
Payee: {transaction.payee}
Amount: {self._format_float(transaction.amount)}
Date: {transaction.date.strftime("%A %B %-d, %Y")}
Category: {category}
- Account: {self.asset_mapping[account_id]}
+ Account: {account_name}
"""
).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: