API Reference

Client

magento.format_datetime(dt: datetime.datetime)

Format a datetime for Magento.

magento.parse_datetime(s: str)

Parse a datetime string from Magento.

magento.client

class magento.client.Magento(token: Optional[str] = None, base_url: Optional[str] = None, scope: Optional[str] = None, logger: Optional[logging.Logger] = None, read_only=False, user_agent=None, *, batch_page_size: Optional[int] = None, **kwargs)

Client for the Magento API.

PAGE_SIZE = 1000

add_product_to_category(category_id: Union[int, str], product_link: Dict[str, Any], **kwargs)

Assign a product to a category.

https://adobe-commerce.redoc.ly/2.4.6-admin/tag/categoriescategoryIdproducts/#operation/PostV1CategoriesCategoryIdProducts

Parameters

• category_id – ID of the category

• product_link – product link. See the Adobe Commerce documentation for the format.

add_products_attribute_option(attribute_code: str, option: Dict[str, str]) → str

Add an option to a products attribute.

https://magento.redoc.ly/2.3.6-admin/#operation/catalogProductAttributeOptionManagementV1AddPost

Parameters

• attribute_code –

• option – dict with label/value keys (mandatory)

Returns

new id

assign_attribute_set_attribute(attribute_set_id: int, attribute_group_id: int, attribute_code: str, sort_order: int = 0, **kwargs)

Assign an attribute to an attribute set.

Parameters

• attribute_set_id – ID of the attribute set.

• attribute_group_id – ID of the attribute group. It must be in the attribute set.

• attribute_code – code of the attribute to add in that attribute group and so in that attribute set.

• sort_order –

• kwargs –

Returns

async_update_products(product_updates: Iterable[Dict[str, Any]])

Update multiple products using the async bulk API.

Example:

>>> Magento().async_update_products([{"sku": "SK123", "name": "Abc"}), {"sku": "SK4", "name": "Def"}])

See https://devdocs.magento.com/guides/v2.4/rest/bulk-endpoints.html

Parameters

product_updates – sequence of product data dicts. They MUST contain an sku key.

Returns

create_category(category: Dict[str, Any], **kwargs)

Create a new category.

create_order_invoice(order_id: Union[int, str], payload: Optional[dict] = None, notify=True)

Create an invoice for an order.

See: * https://devdocs.magento.com/guides/v2.4/rest/tutorials/orders/order-create-invoice.html * https://www.rakeshjesadiya.com/create-invoice-using-rest-api-magento-2/

Parameters

• order_id – Order id.

• payload – payload to send to the API.

• notify – if True (default), notify the client. This is overridden by payload.

Returns

delete_attribute(attribute_code: str, **kwargs)

delete_default_source_items()

Delete all source items that have a source_code=default.

Returns

requests.Response object if there are default source items, None otherwise.

delete_product(sku: str, skip_missing=False, throw=True, **kwargs) → bool

Delete a product given its SKU.

Parameters

• sku –

• skip_missing – if true, don’t raise if the product is missing, and return False.

• throw – throw on error response

• kwargs – keyword arguments passed to all underlying methods.

Returns

a boolean indicating success.

delete_product_media(sku: str, media_id: Union[int, str], throw=False)

Delete a media associated with a product.

Parameters

• sku – SKU of the product

• media_id –

• throw –

Returns

delete_products_attribute_option(attribute_code: str, option_id: Union[int, str]) → bool

Remove an option to a products attribute.

Parameters

• attribute_code –

• option_id –

Returns

boolean

delete_source_items(source_items: Iterable[Dict[str, Any]], throw=True, **kwargs)

Delete a sequence of source-items. Only the SKU and the source_code are used. Note: Magento returns an error if this is called with empty source_items.

Parameters

• source_items –

• throw –

• kwargs – keyword arguments passed to the underlying POST call.

Returns

requests.Response object

delete_special_prices(special_prices: Sequence[Dict[str, Any]])

Delete a sequence of special prices.

delete_special_prices_by_sku(skus: Sequence[str])

Equivalent of delete_special_prices(get_special_prices(skus)).

get_attribute_set_attributes(attribute_set_id: int, **kwargs)

Get all attributes for the given attribute set id.

get_attribute_sets(query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all attribute sets (generator).

get_base_prices(skus: Sequence[str]) → List[Dict[str, Any]]

Get base prices for a sequence of SKUs.

get_bulk_status(bulk_uuid: str) → Dict[str, Any]

Get the status of an async/bulk operation.

get_carts(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all carts (generator).

get_categories(query: Optional[Dict[str, Any]] = None, path_prefix: Optional[str] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Yield all categories.

Parameters

• path_prefix – optional path prefix for the categories. Example: "1/2" for all categories whose path is "1/2/...", including "1/2" itself. Use "1/2/" to exclude "1/2" from the returned categories.

• query – optional query. This overrides path_prefix.

• limit – optional limit

get_category(category_id: Union[int, str]) → Optional[Dict[str, Any]]

Return a category given its id.

get_category_by_name(name: str) → Optional[Dict[str, Any]]

Return the first category with the given name.

Parameters

name – exact name of the category

Returns

get_category_products(category_id: Union[int, str], **kwargs) → List[Dict[str, Any]]

Get products assigned to a category.

https://adobe-commerce.redoc.ly/2.4.6-admin/tag/categoriescategoryIdproducts#operation/GetV1CategoriesCategoryIdProducts

Example:

{‘sku’: ‘MYSKU123’, ‘position’: 2, ‘category_id’: ‘17’}

get_cms_blocks(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all CMS blocks (generator).

get_cms_pages(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all CMS pages (generator).

get_coupons(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all coupons (generator).

get_credit_memos(query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all credit memos (generator).

get_current_store_group_id(*, skip_store_groups=False) → int

Get the current store group id for the current scope. This is not part of Magento API.

Parameters

skip_store_groups – if True, assume the current scope is not already a store group.

get_customer(customer_id: int) → dict

Return a single customer.

get_customer_groups(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all customer groups (generator).

get_customers(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all customers (generator).

get_invoice(invoice_id: int) → Dict[str, Any]

get_invoice_by_increment_id(increment_id: str) → Optional[Dict[str, Any]]

get_invoices(query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all invoices (generator).

get_last_orders(limit=10) → List[Dict[str, Any]]

Return a list of the last orders (default: 10).

get_manufacturers()

Shortcut for .get_products_attribute_options(“manufacturer”).

get_modules(query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all enabled modules (generator).

get_order(order_id: str, throw=True) → Optional[Dict[str, Any]]

Get an order given its (entity) id.

get_order_by_increment_id(increment_id: str) → Optional[Dict[str, Any]]

Get an order given its increment id. Return None if the order doesn’t exist.

get_order_invoices(order_id: Union[int, str])

Get invoices for the given order id.

get_order_shipments(order_id: Union[int, str])

Get shipments for the given order id.

get_orders(*, status: Optional[str] = None, status_condition_type: Optional[str] = None, limit=- 1, query: Optional[Dict[str, Any]] = None, retry=0) → Iterator[Dict[str, Any]]

Return a generator of all orders with this status up to the limit.

Parameters

• status – order status, e.g. “awaiting_shipping”. This overrides query.

• status_condition_type – condition type to use for the status. Default is “eq”. This has no effect if status is not given.

• limit – maximum number of orders to yield (default: no limit).

• query – optional query.

• retry – max retries count

Returns

generator of orders

get_orders_items(*, sku: Optional[str] = None, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs)

Return orders items.

Parameters

• sku – filter orders items on SKU. This is a shortcut for query=make_field_value_query("sku", sku).

• query – optional query. This take precedence over sku.

• limit –

Returns

get_paginated(path: str, *, query: Optional[Dict[str, Any]] = None, limit=- 1, retry=0)

Get a paginated API path.

Parameters

• path –

• query –

• limit – -1 for no limit

• retry –

Returns

get_product(sku: str) → Optional[Dict[str, Any]]

Get a single product. Return None if it doesn’t exist.

Parameters

sku – SKU of the product

Returns

get_product_by_id(product_id: int) → Optional[Dict[str, Any]]

Get a product given its id. Return None if the product doesn’t exist.

Parameters

product_id – ID of the product

Returns

get_product_by_query(query: Optional[Dict[str, Any]], *, expect_one=True) → Optional[Dict[str, Any]]

Get a product with a custom query. Return None if the query doesn’t return match any product, and raise an exception if it returns more than one, unless expect_one is set to False.

Parameters

• query –

• expect_one – if True (the default), raise an exception if the query returns more than one result.

Returns

get_product_media(sku: str, media_id: Union[int, str]) → Dict[str, Any]

Return a gallery entry.

Parameters

• sku – SKU of the product.

• media_id –

Returns

get_product_medias(sku: str) → Sequence[Dict[str, Any]]

Get the list of gallery entries associated with the given product.

Parameters

sku – SKU of the product.

Returns

get_product_stock_item(sku: str) → Dict[str, Any]

Get the stock item for an SKU.

get_product_stock_status(sku: str) → Dict[str, Any]

Get stock status for an SKU.

get_products(limit=- 1, query: Optional[Dict[str, Any]] = None, retry=0) → Iterator[Dict[str, Any]]

Return a generator of all products.

Parameters

• limit – -1 for unlimited.

• query –

• retry –

Returns

get_products_attribute_options(attribute_code: str) → Sequence[Dict[str, str]]

Get all options for a products attribute.

Parameters

attribute_code –

Returns

sequence of option dicts.

get_products_types() → Sequence[Dict[str, Any]]

Get available product types.

get_root_category_id() → int

Get the root category id of the current scope. This is not part of Magento API.

get_sales_rules(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all sales rules (generator).

get_shipments(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Return shipments (generator).

get_source(source_code: str) → Optional[Dict[str, Any]]

Get a single source, or None if it doesn’t exist.

https://adobe-commerce.redoc.ly/2.4.6-admin/tag/inventorysourcessourceCode#operation/GetV1InventorySourcesSourceCode

get_source_items(source_code: Optional[str] = None, sku: Optional[str] = None, *, skus: Optional[Iterable[str]] = None, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Return a generator of all source items.

Parameters

• source_code – optional source_code to filter on. This takes precedence over the query parameter.

• sku – optional SKU to filter on. This takes precedence over the query and the skus parameter.

• skus – optional SKUs list to filter on. This takes precedence of the query parameter.

• query – optional query.

• limit – -1 for unlimited.

Returns

get_sources(query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all sources.

https://adobe-commerce.redoc.ly/2.4.6-admin/tag/inventorysources#operation/GetV1InventorySources

get_special_prices(skus: Sequence[str]) → List[Dict[str, Any]]

Get special prices for a sequence of SKUs.

get_stock_source_links(query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

get_store_configs(store_codes: Optional[List[str]] = None) → Iterable[Dict[str, Any]]

get_store_groups() → Iterable[Dict[str, Any]]

get_store_views() → Iterable[Dict[str, Any]]

get_tax_classes(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all tax classes (generator).

get_tax_rates(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all tax rates (generator).

get_tax_rules(*, query: Optional[Dict[str, Any]] = None, limit=- 1, **kwargs) → Iterable[Dict[str, Any]]

Get all tax rules (generator).

get_websites() → Iterable[Dict[str, Any]]

hold_order(order_id: str, **kwargs)

Hold an order. This is the opposite of unhold_order.

Parameters

order_id – order id (not increment id)

link_child_product(parent_sku: str, child_sku: str, **kwargs) → requests.models.Response

Link two products, one as the parent of the other.

Parameters

• parent_sku – SKU of the parent product

• child_sku – SKU of the child product

Returns

requests.Response object

remove_attribute_set_attribute(attribute_set_id: int, attribute_code: str, **kwargs)

remove_category(category_id: Union[int, str], **kwargs)

Remove a category.

https://adobe-commerce.redoc.ly/2.4.6-admin/tag/categoriescategoryId/#operation/DeleteV1CategoriesCategoryId

remove_product_from_category(category_id: Union[int, str], sku: str, **kwargs)

Remove a product from a category.

https://adobe-commerce.redoc.ly/2.4.6-admin/tag/categoriescategoryIdproductssku/#operation/DeleteV1CategoriesCategoryIdProductsSku

Parameters

• category_id – ID of the category

• sku – SKU of the product

request_api(method: str, path: str, *args, async_bulk=False, throw=False, retry=0, **kwargs)

Equivalent of .request() that prefixes the path with the base API URL.

Parameters

• method – HTTP method

• path – API path. This must start with “/V1/”

• args – arguments passed to .request()

• async_bulk – if True, use the “/async/bulk” prefix. https://devdocs.magento.com/guides/v2.3/rest/bulk-endpoints.html

• throw – if True, raise an exception if the response is an error

• retry – if non-zero, retry the request that many times if there is an error, sleeping 10s between each request.

• kwargs – keyword arguments passed to .request()

Returns

save_attribute(attribute: Dict[str, Any], *, with_defaults=True, throw=True, **kwargs) → Dict[str, Any]

save_base_prices(prices: Sequence[Dict[str, Any]])

Save base prices.

Example:

>>> self.save_base_prices([{"price": 3.14, "sku": "W1033", "store_id": 0}])

Parameters

prices – base prices to save.

Returns

requests.Response object

save_configurable_product_option(sku: str, option: Dict[str, Any], throw=False)

Save a configurable product option.

Parameters

• sku – SKU of the product

• option – option to save

• throw –

Returns

requests.Response object

save_order(order: Dict[str, Any])

Save an order.

save_product(product, *, save_options: Optional[bool] = None) → Dict[str, Any]

Save a product.

Parameters

• product – product to save (can be partial).

• save_options – set the saveOptions attribute.

Returns

save_product_media(sku: str, media_entry: Dict[str, Any])

Save a product media.

save_source(source: Dict[str, Any])

Save a source.

https://adobe-commerce.redoc.ly/2.4.6-admin/tag/inventorysources/#operation/PostV1InventorySources

save_source_items(source_items: Sequence[Dict[str, Any]])

Save a sequence of source-items. Return None if the sequence is empty.

Parameters

source_items –

Returns

save_special_prices(special_prices: Sequence[Dict[str, Any]])

Save a sequence of special prices.

Example:

>>> price_from = "2022-01-01 00:00:00"
>>> price_to = "2022-01-31 23:59:59"
>>> special_price = {"store_id": 0, "sku": "W1033", "price": 2.99,                                  "price_from": price_from, "price_to": price_to}
>>> self.save_special_prices([special_price])

Parameters

special_prices – Special prices to save.

Returns

set_order_status(order: Dict[str, Any], status: str, *, external_order_id: Optional[str] = None)

Change the status of an order, and optionally set its ext_order_id. This is a convenient wrapper around save_order.

Parameters

• order – order payload

• status – new status

• external_order_id – optional external order id

Returns

set_product_stock_item(sku: str, quantity: int, is_in_stock=1)

Parameters

• sku –

• quantity –

• is_in_stock –

Returns

requests.Response

ship_order(order_id: Union[int, str], payload: Dict[str, Any])

Ship an order.

unhold_order(order_id: str, **kwargs)

Un-hold an order. This is the opposite of hold_order.

Parameters

order_id – order id (not increment id)

unlink_child_product(parent_sku: str, child_sku: str, **kwargs) → requests.models.Response

Opposite of link_child_product().

Parameters

• parent_sku – SKU of the parent product

• child_sku – SKU of the child product

Returns

requests.Response object

update_category(category_id: Union[int, str], category_data: Dict[str, Any]) → Dict[str, Any]

Update a category.

Parameters

• category_id –

• category_data – (partial) category data to update

Returns

updated category

update_product(sku: str, product: Dict[str, Any], *, save_options: Optional[bool] = None) → Dict[str, Any]

Update a product.

Example:

>>> Magento().update_product("SK1234", {"name": "My New Name"})

To update the SKU of a product, pass its id along the new SKU and set save_options=True:

>>> Magento().update_product("old-sku", {"id": 123, "sku": "new-sku"}, save_options=True)

Parameters

• sku – SKU of the product to update

• product – (partial) product data to update

• save_options – set the saveOptions attribute.

Returns

updated product

Helpers

magento.attributes

Custom attributes utilities.

magento.attributes.get_boolean_custom_attribute(item: dict, attribute_code: str) → Optional[bool]

Equivalent of get_custom_attribute(item, attribute_code, coerce_as=bool) with proper typing.

magento.attributes.get_custom_attribute(item: dict, attribute_code: str, coerce_as: Callable[[str], magento.attributes.T]) → Union[None, magento.attributes.T, List[magento.attributes.T]]

magento.attributes.get_custom_attribute(item: dict, attribute_code: str) → Union[None, str, List[str]]

Get a custom attribute from an item given its code.

For example:

>>> get_custom_attribute(..., "my_custom_attribute")
"0"

>>> get_custom_attribute(..., "my_custom_attribute", bool)
False

Parameters

• item –

• attribute_code –

• coerce_as – optional callable that is called on the attribute value if it’s set. This is useful to circumvent Magento’s limitation where all attribute values are strings.

Returns

attribute value or None.

magento.attributes.get_custom_attributes_dict(item: Dict[str, Any]) → OrderedDict[str, Union[Sequence[str], str]]

Get all custom attributes from an item as an ordered dict of code->value.

magento.attributes.serialize_attribute_value(value: Optional[Union[str, int, float, bool]], force_none=False)

Serialize a value to be stored in a Magento attribute.

magento.attributes.set_custom_attribute(item: dict, attribute_code: str, attribute_value: Optional[Union[str, int, float, bool]], *, force_none=False)

Set a custom attribute in an item dict.

For example:

>>> set_custom_attribute({}, "my_custom_attribute", 42)
>>> set_custom_attribute({}, "my_custom_attribute", False)

Parameters

• item – item dict. It’s modified in-place.

• attribute_code –

• attribute_value –

• force_none – by default, the attribute value None is serialized as an empty string. Setting this parameter to True forces this attribute value to None instead. This can be used to delete attributes.

Returns

the modified item dict.

magento.attributes.set_custom_attributes(item: dict, attributes: Iterable[Tuple[str, Optional[Union[str, int, float, bool]]]], *, force_none=False)

Set custom attributes in an item dict. Like set_custom_attribute but with an iterable of attributes.

Parameters

• item – item dict. It’s modified in-place.

• attributes – iterable of label/value attribute tuples

• force_none – see set_custom_attribute for usage.

Returns

the modified item dict.

magento.batches

class magento.batches.BatchGetter(getter: Callable[[...], Iterable[magento.batches.T]], key_field: str, keys: Iterable, batch_size=50)

Base class to create generators of Magento items that can be retrieved from the API using queries. This retrieves items in batches but can be iterated on like any iterator.

class magento.batches.BatchSaver(client: magento.client.Magento, api_path: str, batch_size=500)

Base class to create context managers for asynchronous batches.

add_item(item_data)

Add an item to the current batch. If it makes the batch large enough, it’s sent to the API and a new empty batch is created.

finalize()

Send the last pending batch (if any). This doesn’t need to be called when the object is used as a context manager.

Returns

a dictionary with the total number of batches and items

send_batch()

Send the current pending batch (if any).

class magento.batches.ProductBatchGetter(client: magento.client.Magento, skus: Iterable[str])

Get a bunch of products from an iterable of SKUs:

>>> products = ProductBatchGetter(Magento(), ["sku1", "sku2", ...])
>>> for product in products:
...     print(product)

class magento.batches.ProductBatchSaver(client: magento.client.Magento, batch_size=500)

Context manager to add products to an asynchronous batch job.

>>> with ProductBatchSaver() as p:
...     for product_data in ...:
...         p.save_product(product_data)

save_product(product_data: dict)

magento.order_helpers

magento.order_helpers.get_order_shipping_address(order: Dict[str, Any]) → Dict[str, Any]

Return the first shipping address of an order. Note the returned dict is a reference, so if you modify it, it modifies the order. Make a copy if you want to modify the address without affecting the order.

magento.order_helpers.is_order_cash_on_delivery(order: Dict[str, Any])

Test if an order is paid with ‘cash-on-delivery’.

magento.order_helpers.is_order_on_hold(order: Dict[str, Any])

Test if an order is on hold.