Получить данные из значений JSON

Я работал над тем, где я хочу улучшить свои знания, чтобы сделать безопасный парсинг JSON. Таким образом, я мог бы в значительной степени сделать уродливый обходной путь и превратить его в попытку, но это не так. Я хотел бы знать, где что-то пошло не так и как я могу улучшить код, с которым работал:

import json
from json import JSONDecodeError

import requests

payload = {
    "name": None,
    "price": None,
    "sizes": None
}

headers = {'origin': 'https://www.aplace.com', 'Content-Type': 'application/json'}
with requests.get("http://api.aplace.com/sv/dam/aaltoili-iso-mehu-gren-red-sand", headers=headers) as response:
    if not response.ok:
        print(f"Requests NOT ok -> {payload}")

    try:
        # They added extra comma to fail the JSON. Fixed it by removing duplicate commas
        doc = json.loads(response.text.replace(",,", ",")).get("productData", {})
    except (JSONDecodeError, KeyError, ValueError) as err:
        print(f"Not able to parse json -> {err} -> Payload: {payload}")
        raise

if doc:

    product_name = doc.get('name', {})
    product_brand = doc.get('brandName', {})
    product_price = doc.get('markets', {}).get('6', {}).get('pricesByPricelist', {}).get('3', {}).get('price', {})

    # Added exception incase the [0] is not able to get any value from 'media'
    try:
        product_image = doc.get('media', {})[0].get('sources', {}).get('full', {}).get('url', {})
    except KeyError:
        product_image = None

    product_size = doc.get('items', {})

    try:

        if product_name and product_brand:
            payload['name'] = f"{product_brand} {product_name}"

        if product_price:
            payload['price'] = product_price.replace('xa0', '')

        if product_image:
            payload['image'] = product_image

        if product_size:

            count = 0
            sizes = []
            sizesWithStock = []
            for value in product_size.values():
                if value.get('stockByMarket', {}).get('6', {}):
                    count += value['stockByMarket']['6']
                    sizes.append(value['name'])
                    sizesWithStock.append(f"{value['name']} - ({value['stockByMarket']['6']})")

            payload['sizes'] = sizes
            payload['sizesWithStock'] = sizesWithStock
            payload['stock'] = count

    except Exception as err:
        print("Create notification if we hit here later on")
        raise

print(f"Finished payload -> {payload}")

Моя цель здесь — охватить все значения, и если мы не находим его внутри json, мы используем операторы if, чтобы увидеть, поймали ли мы значение или нет.

Что меня раздражает:

1. Длинные вложенные диктовки, но я не уверен, есть ли другой способ, есть ли лучший способ?
2. try-except for product_image, потому что это список, и я просто хочу взять первое в списке, можно ли пропустить try-except?
3. Если произойдет непредвиденная ошибка, распечатайте исключение, но правильно ли я это делаю?

Надеюсь, отсюда я смогу получить новые знания 🙂

• Вам нужно избавиться от своего антипаттерна «полезной нагрузки». Создаваемый вами словарь не является полезной нагрузкой; это просто мешок с нетипизированными, неорганизованными данными. Если не ООП, то хотя бы поместите это в обычные переменные; но избегайте словаря, если вы на самом деле не делаете что-то, что требует словаря — например, сохранение в файл JSON или отправка во вторичный веб-сервис.
• Напишите универсальную функцию, которая принимает название продукта, а не просто кучу глобального кода.
• Рассмотрите возможность добавления подсказок типа PEP484
• Использовать response.raise_for_status вместо проверки ok себя, особенно потому, что вы заботитесь о правильной обработке исключений
• Ваша глобальная замена запятой небезопасна. Я предполагаю, что вместо Они добавили дополнительную запятую, чтобы не выполнить JSON, они просто ошиблись — никогда не приписывайте злобе то, что вы можете приписать невежеству. Попытка заменить это более узко.
• Вы, кажется, хотите использовать .get('x', {}) как волшебную пулю, не понимая полностью, что она делает, потому что вы использовали ее в некоторых местах, где она вообще не подходит, например .get('price', {}). Ты сделаешь нет хотите установить цену по умолчанию для словаря.
• Подумайте о том, чтобы обернуть исключения в свой собственный тип и перехватить это на верхнем уровне
• По возможности используйте магические константы, например 6 в константе или избегайте этого, выполнив поиск по запросу «то, что вы действительно хотите»; например, строка с конкретной валютой.

Предложенный

import json
from decimal import Decimal
from json import JSONDecodeError
from pprint import pformat
from typing import Dict, Any, Optional

import requests

MARKET_ID = '6'
CURRENCY = 'SEK'


class APlaceError(Exception):
    pass


def get_api(product: str) -> Dict[str, Any]:
    headers = {
        'Origin': 'https://www.aplace.com',
        'Accept': 'application/json',
    }
    try:
        with requests.get(f'http://api.aplace.com/sv/dam/{product}', headers=headers) as response:
            response.raise_for_status()
            payload = response.text.replace(',,"cached"', ',"cached"')
        doc = json.loads(payload)
    except IOError as err:
        raise APlaceError('Unable to get response from APlace server') from err
    except JSONDecodeError as err:
        raise APlaceError('Unable to parse JSON') from err

    if doc.get('content', {}).get('is404'):
        raise APlaceError(f'Product "{product}" not found')

    return doc


class Product:
    def __init__(self, doc: Dict[str, Any]):
        try:
            data = doc['productData']
        except KeyError as e:
            raise APlaceError('Empty response') from e

        product = data.get('product')
        if product is None:
            self.id: Optional[int] = None
        else:
            self.id = int(product)

        self.uri: Optional[str] = data.get('uri')
        self.name: Optional[str] = data.get('name')
        self.brand_name: Optional[str] = data.get('brandName')
        self.sku: Optional[str] = data.get('sku')
        # etc.

        try:
            prices = next(
                price
                for price in data.get('markets', {})
                .get(MARKET_ID, {})
                .get('pricesByPricelist', {})
                .values()
                if price.get('price', '').endswith(CURRENCY)
            )
            price = prices.get('priceAsNumber')
            if price is None:
                self.price: Optional[float] = None
            else:
                self.price = Decimal(price)
            self.price_with_currency: Optional[str] = prices.get('price')
        except StopIteration:
            self.price = None
            self.price_with_currency = None

        # Added exception incase the [0] is not able to get any value from 'media'
        try:
            self.image_url: Optional[str] = next(
                medium.get('sources', {}).get('full', {}).get('url')
                for medium in data.get('media', [])
            )
        except StopIteration:
            self.image_url = None

        self.stock_by_size = {
            item.get('name'): item.get('stockByMarket').get(MARKET_ID, 0)
            for item in data.get('items', {}).values()
        }

    def __str__(self):
        return (
            f'Brand: {self.brand_name}n'
            f'Name: {self.name}n'
            f'Price: {self.price_with_currency}n'
            f'Image URL: {self.image_url}n'
            f'Stock for each size: {pformat(self.stock_by_size)}n'
        )


def main():
    try:
        prod = Product(get_api('aaltoili-iso-mehu-gren-red-sand'))
        print(prod)
    except APlaceError as e:
        print(f'Failed to query product: {e}')


if __name__ == '__main__':
    main()

1. Вам нужны функции.
   Функции помогают разделить ваш код и упростить понимание того, как все сочетается друг с другом.

   У вас должна быть функция для:

   1. Получение страницы.

   2. Разбор JSON с вашим полезным сообщением об ошибке.

   3. Получение полезной нагрузки.

   4. А main функция для вызова всех остальных функций.

      Мы должны вызывать основную функцию только в том случае, если скрипт является точкой входа. Мы можем сделать это с помощью следующего кода:

      if __name__ == "__main__":
          main()
      

2. Вы должны добавить функцию пешеходных дорожек.

   product_price = doc.get('markets', {}).get('6', {}).get('pricesByPricelist', {}).get('3', {}).get('price', {})
   

   # Added exception incase the [0] is not able to get any value from 'media'
   try:
       product_image = doc.get('media', {})[0].get('sources', {}).get('full', {}).get('url', {})
   except KeyError:
       product_image = None
   

   Все было бы намного проще, если бы мы могли просто сделать:

   price = get(data, "markets", "6", "pricesByPricelist", "3", "price")
   image = get(data, "media", 0, "sources", "full", "url")
   

   Построить Гет несложно.

   1. Берем объект, путь и значение по умолчанию.
   2. Для каждого сегмента пути мы индексируем объект.
   3. Если у объекта нет значения (__getitem__ поднимает LookupError) возвращаем значение по умолчанию.
   4. Значение по умолчанию должно вызывать ошибку.
      Мы можем гарантировать, что все, что пользователь передаст функции, будет возвращено. Создав новый экземпляр объекта, мы можем затем проверить, есть ли по умолчанию is экземпляр.
   5. Для удобства мы можем построить get функция, которая по умолчанию None.

   _SENTINEL = object()
   
   def walk(obj, path, default=_SENTINEL):
       try:
           for segment in path:
               obj = obj[segment]
           return obj
       except LookupError:
           if default is _SENTINEL:
               raise LookupError(f"couldn't walk path; {path}") from None
           else:
               return default
   
   def get(obj, *path):
       return walk(obj, path, default=None)
   

3. Я настоятельно рекомендую использовать исключения для потока управления в Python.

   Мы можем создать подкласс BaseException чтобы сделать исключение, которое выводится на печать пользователю, а затем выходит из системы. Обернув main() позвонить в try мы можем проверить, возникает ли наша настраиваемая ошибка print, а затем выйти со статусом 1.

   try:
       main()
   except PrettyExitException as e:
       print(e)
       raise SystemExit(1) from None
   

import json
from json import JSONDecodeError

import requests

_SENTINEL = object()


class PrettyExitException(BaseException):
    pass


def get_page():
    headers = {'origin': 'https://www.aplace.com', 'Content-Type': 'application/json'}
    with requests.get("http://api.aplace.com/sv/dam/aaltoili-iso-mehu-gren-red-sand", headers=headers) as response:
        if not response.ok:
            raise PrettyExitException(f"Requests NOT ok -> {payload}")
        return response.text


def parse_json(raw_json):
    try:
        return json.loads(raw_json)
    except (JSONDecodeError, KeyError, ValueError) as err:
        raise PrettyExitException(f"Not able to parse JSON -> {err} -> Payload: {payload}") from None


def walk(obj, path, default=_SENTINEL):
    try:
        for segment in path:
            obj = obj[segment]
        return obj
    except LookupError:
        if default is _SENTINEL:
            raise LookupError(f"couldn't walk path; {path}") from None
        else:
            return default


def get(obj, *path):
    return walk(obj, path, default=None)


def build_payload(data):
    name = get(data, "name")
    brand = get(data, "brandName")
    price = get(data, "markets", "6", "pricesByPricelist", "3", "price")
    image = get(data, "media", 0, "sources", "full", "url")
    size = get(data, "items")
    count = 0
    sizes = []
    sizes_with_stock = []

    if name is not None and brand is not None:
        name = f"{brand} {name}"
    else:
        name = None

    if price is not None:
        price = price.replace("xa0", "")

    if size is not None:
        for value in size.values():
            stock = get(value, "stockByMarket", "6")
            if stock is not None:
                _name = get(value, "name")
                count += stock
                sizes.append(_name)
                sizes_with_stock.append(f"{_name} - ({stock})")

    return {
        "name": name,
        "price": price,
        "image": image,
        "sizes": sizes,
        "sizesWithStock": sizes_with_stock,
        "stock": count,
    }


def main():
    raw_data = get_page()
    data = parse_json(raw_data.replace(",,", ","))
    payload = build_payload(data.get("productData", {}))
    print(f"Finished payload -> {payload}")


if __name__ == "__main__":
    try:
        main()
    except PrettyExitException as e:
        print(e)
        raise SystemExit(1) from None