Add param to output as json-ld

Author: leonghui

app.py

@@ -1,12 +1,12 @@
 from logging import Logger, getLogger
+from typing import Any
 
-from curl_cffi import Response, Session
+from curl_cffi import Response as CurlResponse, Session
 from fastapi import Depends, FastAPI, HTTPException
-from fastapi.responses import JSONResponse
+from fastapi.responses import HTMLResponse, JSONResponse, Response
 
 from config.constants import DEFAULT_USER_AGENT
-from models.extended_feed import ExtendedJsonFeedItem, ExtendedJsonFeedTopLevel
-from models.feed import JsonFeedItem, JsonFeedTopLevel
+from models.feed import JsonFeedTopLevel
 from models.query import (
     AmazonAsinQuery,
     AmazonKeywordQuery,
@@ -18,6 +18,7 @@
 from parsers.item_parser import parse_item_details
 from parsers.search_parser import parse_search_results
 from services.item_generator import get_top_level_feed
+from services.ld_generator import get_html
 from services.response_handler import get_response
 from services.url_builder import get_dimension_url, get_search_url
 
@@ -26,103 +27,84 @@
 
 
 class AmazonFeedGenerator:
-    def __init__(self) -> None:
-        """
-        Initialize the Amazon Feed Generator with configuration and setup.
-        """
-        # Setup can be done here if needed
-
     def create_query_config(self) -> QueryConfig:
         return QueryConfig(
             session=Session(),
             logger=logger,
             useragent=DEFAULT_USER_AGENT,
         )
 
+    def process_query(
+        self,
+        params: QueryParams,
+        query_class: type[AmazonKeywordQuery | AmazonAsinQuery],
+        url_builder_func,
+        parser_func,
+    ) -> Response:
+        try:
+            config: QueryConfig = self.create_query_config()
+
+            query: AmazonAsinQuery | AmazonKeywordQuery = query_class(
+                status=QueryStatus(),
+                query_str=params.q,
+                locale=convert_to_locale(value=params.country),
+                min_price=params.min_price,
+                max_price=params.max_price,
+                jsonld=params.jsonld,
+                config=config,
+            )
 
-feed_generator: AmazonFeedGenerator = AmazonFeedGenerator()
-
-
-@app.get(path="/")
-@app.get(path="/query")
-async def keyword_search(params: QueryParams = Depends()) -> JSONResponse:
-    """
-    Handle keyword search requests.
-    """
-    try:
-        config: QueryConfig = feed_generator.create_query_config()
-
-        query: AmazonKeywordQuery = AmazonKeywordQuery(
-            status=QueryStatus(),
-            query_str=params.q,
-            locale=convert_to_locale(value=params.country),
-            min_price=params.min_price,
-            max_price=params.max_price,
-            strict=params.strict,
-            config=config,
-        )
-
-        base_url: str = f"https://{query.locale.domain}"
-        search_url: str = get_search_url(base_url, query)
-
-        response: Response | JSONResponse = get_response(url=search_url, query=query)
+            base_url: str = f"https://{query.locale.domain}"
+            search_url: Any = url_builder_func(base_url, query)
 
-        if isinstance(response, Response):
-            feed_items: list[JsonFeedItem] = parse_search_results(
-                response.content, query, base_url
+            response: CurlResponse | JSONResponse = get_response(
+                url=search_url, query=query
             )
 
-            json_feed: JsonFeedTopLevel = get_top_level_feed(
-                base_url, query, feed_items
-            )
+            if isinstance(response, CurlResponse):
+                feed_items: list = parser_func(
+                    response.content or response.json(), query, base_url
+                )
 
-            return JSONResponse(content=json_feed.model_dump(exclude_none=True))
-        else:
-            return response
-
-    except Exception as e:
-        logger.error(msg=f"Keyword search error: {e}")
-        raise HTTPException(status_code=500, detail=f"Keyword search error: {e}")
+                if params.jsonld:
+                    html_text: str = get_html(feed_items)
+                    return HTMLResponse(content=html_text)
+                else:
+                    json_feed: JsonFeedTopLevel = get_top_level_feed(
+                        base_url, query, feed_items
+                    )
+                    return JSONResponse(content=json_feed.model_dump(exclude_none=True))
 
+            return response
 
-@app.get(path="/asin")
-async def asin_lookup(params: QueryParams = Depends()) -> JSONResponse:
-    """
-    Handle ASIN lookup requests.
-    """
-    try:
-        config: QueryConfig = feed_generator.create_query_config()
-
-        query: AmazonAsinQuery = AmazonAsinQuery(
-            status=QueryStatus(),
-            query_str=params.q,
-            locale=convert_to_locale(value=params.country),
-            min_price=params.min_price,
-            max_price=params.max_price,
-            config=config,
-        )
+        except Exception as e:
+            error_msg: str = f"{'Keyword' if query_class is AmazonKeywordQuery else 'ASIN'} lookup error: {e}"
+            logger.error(msg=error_msg)
+            raise HTTPException(status_code=500, detail=error_msg)
 
-        base_url: str = f"https://{query.locale.domain}"
-        search_url: str = get_dimension_url(query)
 
-        response: Response | JSONResponse = get_response(url=search_url, query=query)
+feed_generator: AmazonFeedGenerator = AmazonFeedGenerator()
 
-        if isinstance(response, Response):
-            feed_items: list[JsonFeedItem | ExtendedJsonFeedItem] = parse_item_details(
-                response.json(), query, base_url
-            )
 
-            json_feed: ExtendedJsonFeedTopLevel = get_top_level_feed(
-                base_url, query, feed_items
-            )
+@app.get(path="/")
+@app.get(path="/query")
+async def keyword_search(params: QueryParams = Depends()) -> Response:
+    return feed_generator.process_query(
+        params,
+        query_class=AmazonKeywordQuery,
+        url_builder_func=get_search_url,
+        parser_func=parse_search_results,
+    )
 
-            return JSONResponse(content=json_feed.model_dump(exclude_none=True))
-        else:
-            return response
 
-    except Exception as e:
-        logger.error(msg=f"ASIN lookup error: {e}")
-        raise HTTPException(status_code=500, detail=f"ASIN lookup error: {e}")
+@app.get(path="/asin")
+async def asin_lookup(params: QueryParams = Depends()) -> Response:
+    return feed_generator.process_query(
+        params,
+        query_class=AmazonAsinQuery,
+        url_builder_func=get_dimension_url,
+        parser_func=parse_item_details,
+    )
 
 
 @app.get(path="/healthcheck")

models/amazon.py

@@ -0,0 +1,8 @@
+from typing import Annotated
+from pydantic import AfterValidator, BaseModel
+
+from models.validators import validate_asin
+
+
+class Asin(BaseModel):
+    id: Annotated[str, AfterValidator(func=validate_asin)]

models/amazon/asin.py

@@ -0,0 +1,40 @@
+from pydantic import BaseModel
+
+
+class AmazonLocale(BaseModel):
+    code: str
+    domain: str
+    currency_sign: str
+    currency_code: str
+
+
+locale_list: list[AmazonLocale] = [
+    AmazonLocale(
+        code="AU", domain="www.amazon.com.au", currency_sign="$", currency_code="AUD"
+    ),
+    AmazonLocale(
+        code="DE", domain="www.amazon.de", currency_sign="€", currency_code="EUR"
+    ),
+    AmazonLocale(
+        code="ES", domain="www.amazon.es", currency_sign="€", currency_code="EUR"
+    ),
+    AmazonLocale(
+        code="FR", domain="www.amazon.fr", currency_sign="€", currency_code="EUR"
+    ),
+    AmazonLocale(
+        code="IT", domain="www.amazon.it", currency_sign="€", currency_code="EUR"
+    ),
+    AmazonLocale(
+        code="SG", domain="www.amazon.sg", currency_sign="S$", currency_code="SGD"
+    ),
+    AmazonLocale(
+        code="UK", domain="www.amazon.co.uk", currency_sign="£", currency_code="GBP"
+    ),
+    AmazonLocale(
+        code="US", domain="www.amazon.com", currency_sign="$", currency_code="USD"
+    ),
+]
+
+default_locale: AmazonLocale = next(
+    locale for locale in locale_list if locale.code == "US"
+)

models/amazon/locale.py

@@ -1,7 +1,10 @@
-from pydantic import ConfigDict, Field, PositiveFloat
+from decimal import Decimal
+
+from pydantic import ConfigDict, Field
 from pydantic.main import BaseModel
 
-from models.feed import JsonFeedItem, JsonFeedTopLevel, SerHttpUrl
+from models.amazon.asin import Asin
+from models.feed import SerHttpUrl
 
 
 class Thing(BaseModel):
@@ -12,24 +15,15 @@ class Thing(BaseModel):
 
 class Offer(Thing):
     type: str = Field(default="Offer", serialization_alias="@type")
-    priceCurrency: str | None
-    price: PositiveFloat | None
+    priceCurrency: str | None = None
+    price: Decimal | None = None
     availability: str | None = "https://schema.org/InStock"
 
 
 class Product(Thing):
     type: str = Field(default="Product", serialization_alias="@type")
     context: str = Field(default="https://schema.org/", serialization_alias="@context")
     name: str | None
-    asin: str | None
+    asin: Asin | None
     image: list[SerHttpUrl] | None
     offers: Offer | None
-
-
-class ExtendedJsonFeedItem(JsonFeedItem):
-    # Product schema encapsulated in <script type="application/ld+json">
-    _linked_data: str | None = None
-
-
-class ExtendedJsonFeedTopLevel(JsonFeedTopLevel):
-    items: list[JsonFeedItem | ExtendedJsonFeedItem]

models/extended_feed.py models/json_ld.pymodels/extended_feed.py renamed to models/json_ld.py

@@ -5,11 +5,10 @@
 from fastapi import Query
 from pydantic import AfterValidator, BaseModel, Field, PositiveFloat
 
-from models.amazon import AmazonLocale, default_locale
+from models.amazon.locale import AmazonLocale, default_locale
 from models.validators import (
     validate_asin,
     validate_country,
-    validate_price_filters,
     validate_query_str,
 )
 
@@ -39,9 +38,8 @@ class _BaseQuery(BaseModel):
     status: QueryStatus
     config: QueryConfig
     query_str: str
-    locale: AmazonLocale = (
-        default_locale
-    )
+    locale: AmazonLocale = default_locale
+    jsonld: bool = False
 
 
 class FilterableQuery(_BaseQuery):
@@ -66,10 +64,7 @@ class QueryParams(BaseModel):
     country: Annotated[str, AfterValidator(func=validate_country)] = Field(
         Query("us", description="Country code")
     )
-    min_price: Annotated[int, AfterValidator(func=validate_price_filters)] | None = (
-        Field(Query(None, description="Minimum price"))
-    )
-    max_price: Annotated[int, AfterValidator(func=validate_price_filters)] | None = (
-        Field(Query(None, description="Maximum price"))
-    )
+    min_price: PositiveFloat | None = Field(Query(None, description="Minimum price"))
+    max_price: PositiveFloat | None = Field(Query(None, description="Maximum price"))
     strict: bool | None = Field(Query(False, description="Strict mode"))
+    jsonld: bool = Field(Query(False, description="Return output as JSON-LD"))

models/query.py

@@ -1,6 +1,6 @@
 import re
 
-from models.amazon import AmazonLocale, locale_list
+from models.amazon.locale import AmazonLocale, locale_list
 
 ASIN_PATTERN = r"^(B[\dA-Z]{9}|\d{9}(X|\d))$"
 
@@ -18,12 +18,6 @@ def convert_to_locale(value: str) -> AmazonLocale:
     return next((locale for locale in locale_list if locale.code == country_code))
 
 
-def validate_price_filters(value: str) -> int:
-    if value and not value.isnumeric():
-        raise ValueError("Invalid price filter")
-    return int(value)
-
-
 def validate_query_str(value: str) -> str:
     if value and not len(value):
         raise ValueError("Invalid query")