Merge branch 'DanielKarp-main'

Incorporate changes from upstream to get caught up

Author: WoosterInitiative

.pre-commit-config.yaml

@@ -1,25 +1,25 @@
 repos:
--   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v5.0.0
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v6.0.0
     hooks:
-    -   id: trailing-whitespace
-    -   id: end-of-file-fixer
-    -   id: check-yaml
-    -   id: check-added-large-files
--   repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.11.4
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-added-large-files
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.12.12
     hooks:
-    -   id: ruff
+      - id: ruff-check
         args: [--fix]
-    -   id: ruff-format
--   repo: https://github.com/astral-sh/uv-pre-commit
-    rev: 0.6.13
+      - id: ruff-format
+  - repo: https://github.com/astral-sh/uv-pre-commit
+    rev: 0.8.15
     hooks:
-    -   id: uv-lock
-    -   id: uv-export
+      - id: uv-lock
+      - id: uv-export
         args: [--quiet]
--   repo: https://github.com/fpgmaas/deptry.git
-    rev: 0.23.0
+  - repo: https://github.com/fpgmaas/deptry.git
+    rev: 0.23.1
     hooks:
-    -   id: deptry
+      - id: deptry
         args: [src]

README.md

@@ -1,3 +1,5 @@
+:bangbang: : Development has moved to [WoosterTech fork](https://github.com/WoosterTech/YNAmazon/). This repository is no longer maintained.
+
 # YNAmazon
 A program to annotate YNAB transactions with Amazon order info
 

pyproject.toml

@@ -16,11 +16,12 @@ dependencies = [
     "pydantic-core>=2.33.1",
     "furl>=2.1.4",
     "requests>=2.30.0",
+    "cache-decorator>=2.2.0",
 ]
 
 [project.optional-dependencies]
 ai = [
-    "openai>=1.12.0",  # OpenAI Python client for AI summarization
+    "openai>=1.12.0", # OpenAI Python client for AI summarization
 ]
 
 [project.urls]
@@ -59,8 +60,8 @@ build-backend = "hatchling.build"
 
 
 [[tool.mypy.overrides]]
-module= ["amazonorders.*"]
+module = ["amazonorders.*"]
 ignore_missing_imports = true
 
-# [tool.deptry]
-# known_first_party = ["ynamazon"]
+[tool.deptry]
+known_first_party = ["ynamazon"]

src/ynamazon/amazon_transactions.py

@@ -1,14 +1,16 @@
 # pyright: reportDeprecated=false
+import os
+import tempfile
 from datetime import date
 from decimal import Decimal
 from typing import Annotated, Union  # ,  Self  # not available python <3.11
 
 from amazonorders.entity.order import Order
 from amazonorders.entity.transaction import Transaction
-from amazonorders.exception import AmazonOrdersAuthError
 from amazonorders.orders import AmazonOrders
 from amazonorders.session import AmazonSession
 from amazonorders.transactions import AmazonTransactions
+from cache_decorator import Cache
 from loguru import logger
 from pydantic import AnyUrl, BaseModel, EmailStr, Field, SecretStr, field_validator
 from rich import print as rprint
@@ -75,93 +77,142 @@ def amazon_session(self) -> AmazonSession:
         )
 
 
-def get_amazon_transactions(
-    order_years: list[int] | None = None,
-    transaction_days: int = 31,
-    configuration: AmazonConfig | None = None,
-) -> list[AmazonTransactionWithOrderInfo]:
-    """Returns a list of transactions with order info.
+class AmazonTransactionRetriever:
+    def __init__(
+        self,
+        amazon_config: AmazonConfig,
+        order_years: list[str] | None = None,
+        transaction_days: int = 31,
+        force_refresh_amazon: bool = False,
+    ):
+        """Initialize an AmazonTransactionRetriever.
 
-    Args:
+        amazon_config (AmazonConfig): Configuration for Amazon, primarily credentials
         order_years (list[int] | None): A list of years to fetch transactions for. `None` for the current year.
-        transaction_days (int): Number of days to fetch transactions for.
-        configuration (AmazonConfig | None): Amazon configuration.
+        transaction_days (int): Number of days to fetch transactions for. Defaults to 31.
+        force_refresh_amazon (bool): Refresh cache by fetching transactions directly from Amazon.
+        """
+        self.amazon_config = amazon_config
+        self.order_years = self.__class__._normalized_years(order_years)
+        self.transaction_days = transaction_days
+        self.force_refresh_amazon = force_refresh_amazon
+
+        # for memoizing the results of method calls
+        self._memo = {}
+
+    def get_amazon_transactions(self) -> list[AmazonTransactionWithOrderInfo]:
+        """Get Amazon transactions linked to orders.
+
+        This method exists as a layer to force caching to work one level below with all relevant parameters considered
+
+        Returns:
+            list[TransactionWithOrderInfo]: A list of transactions with order info
+        """
+        return self._get_amazon_transactions(
+            order_years=self.order_years,
+            transaction_days=self.transaction_days,
+            amazon_config=self.amazon_config,
+            use_cache=not self.force_refresh_amazon,
+        )
 
-    Returns:
-        list[TransactionWithOrderInfo]: A list of transactions with order info
-    """
-    if configuration is None:
-        configuration = AmazonConfig()
-    amazon_session = configuration.amazon_session()
-    try:
-        logger.debug("Logging in to Amazon session...")
-        logger.debug(f"Session debug mode: {amazon_session.debug}")  # pyright: ignore[reportUnknownMemberType, reportAttributeAccessIssue]
-        amazon_session.login()
-    except AmazonOrdersAuthError as e:
-        logger.error(f"Failed to authenticate Amazon session: {e}")
-        raise
+    @Cache(
+        validity_duration="2h",
+        enable_cache_arg_name="use_cache",
+        cache_path=os.path.join(
+            tempfile.gettempdir(),
+            "ynamazon",
+            "amazon_transactions_get_amazon_transactions_{_hash}.pkl",
+        ),
+    )
+    def _get_amazon_transactions(
+        self,
+        order_years: list[str],
+        transaction_days: int,
+        amazon_config: AmazonConfig,
+    ) -> list[AmazonTransactionWithOrderInfo]:
+        orders_dict = {order.order_number: order for order in self._amazon_orders()}
+
+        amazon_transactions = self._amazon_transactions()
+
+        amazon_transaction_with_order_details: list[AmazonTransactionWithOrderInfo] = []
+        for transaction in amazon_transactions:
+            try:
+                amazon_transaction_with_order_details.append(
+                    AmazonTransactionWithOrderInfo.from_transaction_and_orders(
+                        orders_dict=orders_dict, transaction=transaction
+                    )
+                )
+            except ValueError:
+                logger.debug(
+                    f"Transaction {transaction.order_number} not found in retrieved orders."
+                )
+                continue
 
-    orders = _fetch_amazon_order_history(session=amazon_session, years=order_years)
-    orders_dict = {order.order_number: order for order in orders}  # pyright: ignore[reportUnknownMemberType, reportAttributeAccessIssue, reportUnknownVariableType]
+        return amazon_transaction_with_order_details
 
-    amazon_transactions = _fetch_sorted_amazon_transactions(
-        transaction_days=transaction_days, amazon_session=amazon_session
-    )
+    def _amazon_orders(self) -> list[Order]:
+        """Returns a list of Amazon orders.
 
-    amazon_transaction_with_order_details: list[AmazonTransactionWithOrderInfo] = []
-    for transaction in amazon_transactions:
-        try:
-            amazon_transaction_with_order_details.append(
-                AmazonTransactionWithOrderInfo.from_transaction_and_orders(
-                    orders_dict=orders_dict,  # pyright: ignore[reportUnknownArgumentType]
-                    transaction=transaction,
-                )
-            )
-        except ValueError:
-            logger.debug(f"Transaction {transaction.order_number} not found in retrieved orders.")  # pyright: ignore[reportUnknownMemberType, reportAttributeAccessIssue]
-            continue
+        Args:
+            years (Sequence[int] | None): A sequence of years to fetch orders for. `None` for the current year.
 
-    return amazon_transaction_with_order_details
+        Returns:
+            list[Order]: A list of Amazon orders.
+        """
+        if "amazon_orders" in self._memo:
+            return self._memo["amazon_orders"]
 
+        amazon_orders = AmazonOrders(self._session())
 
-def _fetch_amazon_order_history(
-    *, session: AmazonSession, years: Union[list[int], None] = None
-) -> list[Order]:
-    """Returns a list of Amazon orders.
+        all_orders: list[Order] = []
+        for year in self.order_years:
+            all_orders.extend(amazon_orders.get_order_history(year=year))
+        all_orders.sort(key=lambda order: order.order_placed_date)
 
-    Args:
-        session (AmazonSession): Amazon session (must be logged in).
-        years (Sequence[int] | None): A sequence of years to fetch orders for. `None` for the current year.
+        self._memo["amazon_orders"] = all_orders
 
-    Returns:
-        list[Order]: A list of Amazon orders.
-    """
-    if not session.is_authenticated:  # pyright: ignore[reportUnknownMemberType, reportAttributeAccessIssue]
-        raise ValueError("Session must be authenticated.")
-    amazon_orders = AmazonOrders(session)
-    if years is None:
-        years = [date.today().year]
-    all_orders: list[Order] = []
-    for year in years:
-        if len(year_str := str(year)) == 2:
-            year_str = "20" + year_str
-        all_orders.extend(amazon_orders.get_order_history(year=year_str))  # pyright: ignore[reportArgumentType]
-    all_orders.sort(key=lambda order: order.order_placed_date)  # pyright: ignore[reportUnknownMemberType, reportUnknownLambdaType, reportAttributeAccessIssue]
-
-    return all_orders
-
-
-def _fetch_sorted_amazon_transactions(
-    *, amazon_session: AmazonSession, transaction_days: int = 31
-) -> list[Transaction]:
-    """Fetches and sorts Amazon transactions."""
-    if not amazon_session.is_authenticated:  # pyright: ignore[reportUnknownMemberType, reportAttributeAccessIssue]
-        raise ValueError("Session must be authenticated.")
-    amazon_transactions = AmazonTransactions(amazon_session=amazon_session).get_transactions(
-        days=transaction_days
-    )
-    amazon_transactions.sort(key=lambda trans: trans.completed_date)  # pyright: ignore[reportUnknownMemberType, reportUnknownLambdaType, reportAttributeAccessIssue]
-    return amazon_transactions
+        return self._memo["amazon_orders"]
+
+    def _amazon_transactions(self) -> list[Transaction]:
+        """Fetches and sorts Amazon transactions."""
+        if "amazon_transactions" in self._memo:
+            return self._memo["amazon_transactions"]
+
+        self._memo["amazon_transactions"] = AmazonTransactions(
+            amazon_session=self._session()
+        ).get_transactions(days=self.transaction_days)
+
+        self._memo["amazon_transactions"].sort(key=lambda trans: trans.completed_date)
+
+        return self._memo["amazon_transactions"]
+
+    def _session(self) -> AmazonSession:
+        if "session" in self._memo:
+            return self._memo["session"]
+
+        amazon_session = self.amazon_config.amazon_session()
+        amazon_session.login()
+
+        if amazon_session.is_authenticated:
+            self._memo["session"] = amazon_session
+            return self._memo["session"]
+
+    @classmethod
+    def _normalized_years(cls, years: list[str] | None = None) -> list[str]:
+        if years is None:
+            return [date.today().year]
+
+        result: list[str] = []
+
+        for year in years:
+            if len(year) == 2:
+                result.append("20" + year)
+            elif len(year) == 4:
+                result.append(year)
+            else:
+                raise ValueError("Year must be specified as 2 or 4 digits (e.g. 21 or 2021)")
+
+        return result
 
 
 def print_amazon_transactions(
@@ -221,5 +272,5 @@ def locate_amazon_transaction_by_amount(
     return None
 
 
-if __name__ == "__main__":
-    print_amazon_transactions(get_amazon_transactions())
+# if __name__ == "__main__":
+# print_amazon_transactions(AmazonTransactionRetriever.new()