The search module

Available Endpoints

The following endpoints are currently wrapped with a Model and SearchQuery subclass

Endpoint	Client Shortcut	SearchQuery Subclass	Model Subclass
orders	Client.orders	OrderSearch	Order
orders/items	Client.order_items	OrderItemSearch	OrderItem
invoices	Client.invoices	InvoiceSearch	Invoice
products	Client.products	ProductSearch	Product
products/attributes	Client.product_attributes	ProductAttributeSearch	ProductAttribute
categories	Client.categories	CategorySearch	Category
customers	Client.customers	CustomerSearch	Customer
endpoint	Client.search('endpoint')	SearchQuery	APIResponse

class magento.search.SearchQuery(endpoint, client, model=<class 'magento.models.model.APIResponse'>)View on GitHub

View Source Code

class SearchQuery:

    """Queries any endpoint that invokes the searchCriteria interface. Parent of all endpoint-specific search classes

    .. tip:: See https://developer.adobe.com/commerce/webapi/rest/use-rest/performing-searches/ for official docs
    """

    def __init__(self, endpoint: str, client: Client, model: Type[Model] = APIResponse):
        """Initialize a SearchQuery object

        :param endpoint: the base search API endpoint (for example, ``orders``)
        :param client: an initialized :class:`~.Client` object
        :param model: the :class:`~.Model` to parse the response data with; uses :class:`~.APIResponse` if not specified
        """
        if not isinstance(client, clients.Client):
            raise TypeError(f'`client` must be of type {clients.Client}')

        #: The :class:`~.Client` to send the search request with
        self.client = client
        #: The endpoint being queried
        self.endpoint = endpoint
        #: :doc:`models` class to wrap the response with
        self.Model = model
        #: The current url for the search request
        self.query = self.client.url_for(endpoint) + '/?'
        #: Restricted fields, from :meth:`~.restrict_fields`
        self.fields = ''
        #: The raw response data, if any
        self._result = {}

    def add_criteria(self, field, value, condition='eq', **kwargs) -> Self:
        """Add criteria to the search query

        :param field: the API response field to search by
        :param value: the value of the field to compare
        :param condition: the comparison condition
        :param kwargs: additional search option arguments (``group`` and ``filter``)
        :returns: the calling SearchQuery object

        .. admonition:: Keyword Argument Options: ``Condition``
           :class: tip

           The ``condition`` argument specifies the condition used to evaluate the attribute value

           * ``"eq"`` (default): matches items for which ``field=value``
           * ``"gt"``: matches items for which ``field>value``
           * ``"lt"``: matches items for which ``field<value``
           * ``"gteq"``: matches items for which ``field>=value``
           * ``"lteq"``: matches items for which ``field<=value``
           * ``"in"``: matches items for which ``field in value.split(",")``

             - Tip: for ``in``, use :meth:`~.by_list` if not building a complex query

           .. admonition:: Example
              :class: example

              ::

               # Search for Orders created in 2023
               >>> orders = api.orders.add_criteria(
               ...     field="created_at",
               ...     value="2023-01-01",
               ...     condition='gteq'
               ... ).execute()

        .. admonition:: Keyword Argument Options: Using Filter Groups
           :class: hint

           ``group`` - filter group number

           ``filter`` - filter number (within the specified filter group)


        *Using Filter Groups*

            Filter groups are filter criteria in the form of { field: value }

                Group 0 Filter 0                        ->      Filter 0
                Group 0 Filter 0 + Group 0 Filter 1     ->      Filter 0 OR Filter 1
                Group 0 Filter 0 + Group 1 Filter 0     ->      Filter 0 AND Filter 0
        """

        options = {
            'condition': condition,
            'group': self.last_group + 1,
            'filter': 0,
        }
        options.update(kwargs)

        criteria = (
                f'searchCriteria[filter_groups][{options["group"]}][filters][{options["filter"]}][field]={field}' +
                f'&searchCriteria[filter_groups][{options["group"]}][filters][{options["filter"]}][value]={value}' +
                f'&searchCriteria[filter_groups][{options["group"]}][filters][{options["filter"]}][condition_type]={options["condition"]}'
        )
        if not self.query.endswith('?'):
            self.query += '&'
        self.query += criteria
        return self

    def restrict_fields(self, fields: Iterable[str]) -> Self:
        """Constrain the API response data to only contain the specified fields

        :param fields: an iterable or comma separated string of fields to include in the response
        :returns: the calling SearchQuery object
        """
        if not isinstance(fields, str):
            if not isinstance(fields, Iterable):
                raise TypeError('"fields" must be a comma separated string or an iterable')
            fields = ','.join(fields)

        if (id_field := self.Model.IDENTIFIER) not in fields:
            fields += f',{id_field}'

        self.fields = f'&fields=items[{fields}]'
        return self

    def execute(self) -> Optional[Model | List[Model]]:
        """Sends the search request using the current :attr:`~.scope` of the :attr:`client`

        .. tip:: Change the :attr:`.Client.scope` to retrieve :attr:`~.result` data
           from different store :attr:`~.views`

        :returns: the search query :attr:`~.result`
        """
        response = self.client.get(self.query + self.fields)
        self.__dict__.pop('result', None)
        self._result = response.json()
        return self.result

    def by_id(self, item_id: Union[int, str]) -> Optional[Model]:
        """Retrieve data for an individual item by its id

        .. note:: The ``id`` field used is different depending on the endpoint being queried

           * Most endpoints use an ``entity_id`` or ``id``
           * The ``orders/items`` endpoint uses ``item_id``
           * The ``products`` endpoint uses ``product_id``,
             but can also be queried :meth:`~ProductSearch.by_sku`

           The :attr:`~.Model.IDENTIFIER` attribute of each :class:`~.Model` contains the appropriate field

        :param item_id: id of the item to retrieve
        """
        self.query = self.query.strip('?') + str(item_id)
        return self.execute()

    def by_list(self, field: str, values: Iterable) -> Optional[Model, List[Model]]:
        """Search for multiple items using an iterable or comma-separated string of field values

        .. admonition:: Examples
           :class: example

           Retrieve :class:`~.Product` with ids from 1 to 10::

            # Values can be a list/tuple/iterable
            >> api.products.by_list('entity_id', range(1,11))

           Search for :class:`~.Order` that are processing, pending, or completed::

            #  Values can be a comma-separated string
            >> api.orders.by_list('status', 'processing,pending,completed')


        :param field: the API response field to search for matches in
        :param values: an iterable or comma separated string of values
        """
        if not isinstance(values, Iterable):
            raise TypeError('`values` must be an iterable')
        if not isinstance(values, str):
            values = ','.join(f'{value}' for value in values)
        return self.add_criteria(
            field=field,
            value=values,
            condition='in'
        ).execute()

    def get_all(self) -> Optional[Model, List[Model]]:
        """Retrieve all items for the given search endpoint.

        .. warning:: Not guaranteed to work with all endpoints.
        """
        return self.since().execute()

    def since(self, sinceDate: str = None) -> Self:
        """Retrieve items for which ``created_at >= sinceDate``

        **Example**::

            # Retrieve products created in 2023
            >> api.products.since('2023-01-01').execute()


        .. tip:: Calling with no arguments retrieves all items

           ::

            # Retrieve all products
            >> api.products.since().execute()

        :param sinceDate: the date for response data to start from
        :return: the calling :class:`~SearchQuery`
        """
        return self.add_criteria(
            field='created_at',
            value=sinceDate,
            condition='gteq',
            group=self.last_group + 1,
        )

    def until(self, toDate: str) -> Self:
        """Retrieve items for which ``created_at <= toDate``

        :param toDate: the date for response data to end at (inclusive)
        :return: the calling :class:`~SearchQuery`
        """
        return self.add_criteria(
            field='created_at',
            value=toDate,
            condition='lteq',
            group=self.last_group + 1,
        )

    @cached_property
    def result(self) -> Optional[Model | List[Model]]:
        """The result of the search query, wrapped by the :class:`~.Model` corresponding to the endpoint

        :returns: the API response as either an individual or list of :class:`~.Model` objects
        """
        result = self.validate_result()
        if result is None:
            return result
        if isinstance(result, list):
            return [self.parse(item) for item in result]
        if isinstance(result, dict):
            return self.parse(result)

    def validate_result(self) -> Optional[Dict | List[Dict]]:
        """Parses the response and returns the actual result data, regardless of search approach"""
        if not self._result:
            return None

        if isinstance(self._result, list):
            return self._result

        if self._result.get('message'):  # Error; logged by Client
            return None

        if len(self._result.keys()) > 3:  # Single item, retrieved by id
            return self._result

        if 'items' in self._result:  # All successful responses with search criteria have `items` key
            items = self._result['items']
            if items:  # Response can still be {'items': None} though
                if len(items) > 1:
                    return items
                return items[0]
            else:
                self.client.logger.info(
                    "No matching {} for this search query".format(self.endpoint)
                )
                return None
        # I have no idea what could've gone wrong, sorry :/
        msg = "Search failed with an unknown error.\nAPI Response: {}".format(self._result)
        raise MagentoError(self.client, msg)

    def parse(self, data: dict) -> Model:
        """Parses the API response with the corresponding :class:`~.Model` object

        :param data: API response data of a single item
        """
        if self.Model is not APIResponse:
            return self.Model(data, self.client)
        return self.Model(data, self.client, self.endpoint)

    def reset(self) -> None:
        """Resets the query and result, allowing the object to be reused"""
        self._result = {}
        self.fields = ''
        self.query = self.client.url_for(self.endpoint) + '/?'
        self.__dict__.pop('result', None)

    @property
    def result_count(self) -> int:
        """Number of items that matched the search criteria"""
        if not self._result or not self.result:
            return 0
        if isinstance(self.result, Model):
            return 1
        return len(self.result)

    @property
    def result_type(self) -> Type:
        """The type of the result"""
        return type(self.result)

    @property
    def last_group(self) -> int:
        """The most recent filter group on the query
        
        :returns: the most recent filter group, or ``-1`` if no criteria has been added
        """
        if self.query.endswith('?'):
            return -1
        return int(re.match(
            r'.*searchCriteria\[filter_groups]\[(\d)]',
            self.query
        ).groups()[-1])

Bases: object

Queries any endpoint that invokes the searchCriteria interface. Parent of all endpoint-specific search classes

Tip

See https://developer.adobe.com/commerce/webapi/rest/use-rest/performing-searches/ for official docs

__init__(endpoint, client, model=<class 'magento.models.model.APIResponse'>)View on GitHub

View Source Code

def __init__(self, endpoint: str, client: Client, model: Type[Model] = APIResponse):
    """Initialize a SearchQuery object

    :param endpoint: the base search API endpoint (for example, ``orders``)
    :param client: an initialized :class:`~.Client` object
    :param model: the :class:`~.Model` to parse the response data with; uses :class:`~.APIResponse` if not specified
    """
    if not isinstance(client, clients.Client):
        raise TypeError(f'`client` must be of type {clients.Client}')

    #: The :class:`~.Client` to send the search request with
    self.client = client
    #: The endpoint being queried
    self.endpoint = endpoint
    #: :doc:`models` class to wrap the response with
    self.Model = model
    #: The current url for the search request
    self.query = self.client.url_for(endpoint) + '/?'
    #: Restricted fields, from :meth:`~.restrict_fields`
    self.fields = ''
    #: The raw response data, if any
    self._result = {}

Initialize a SearchQuery object

Parameters

• endpoint (str) – the base search API endpoint (for example, orders)

• client (Client) – an initialized Client object

• model (Type[Model]) – the Model to parse the response data with; uses APIResponse if not specified

client

The Client to send the search request with

endpoint

The endpoint being queried

Model

The magento.models subpackage class to wrap the response with

query

The current url for the search request

fields

Restricted fields, from restrict_fields()

add_criteria(field, value, condition='eq', **kwargs)View on GitHub

View Source Code

def add_criteria(self, field, value, condition='eq', **kwargs) -> Self:
    """Add criteria to the search query

    :param field: the API response field to search by
    :param value: the value of the field to compare
    :param condition: the comparison condition
    :param kwargs: additional search option arguments (``group`` and ``filter``)
    :returns: the calling SearchQuery object

    .. admonition:: Keyword Argument Options: ``Condition``
       :class: tip

       The ``condition`` argument specifies the condition used to evaluate the attribute value

       * ``"eq"`` (default): matches items for which ``field=value``
       * ``"gt"``: matches items for which ``field>value``
       * ``"lt"``: matches items for which ``field<value``
       * ``"gteq"``: matches items for which ``field>=value``
       * ``"lteq"``: matches items for which ``field<=value``
       * ``"in"``: matches items for which ``field in value.split(",")``

         - Tip: for ``in``, use :meth:`~.by_list` if not building a complex query

       .. admonition:: Example
          :class: example

          ::

           # Search for Orders created in 2023
           >>> orders = api.orders.add_criteria(
           ...     field="created_at",
           ...     value="2023-01-01",
           ...     condition='gteq'
           ... ).execute()

    .. admonition:: Keyword Argument Options: Using Filter Groups
       :class: hint

       ``group`` - filter group number

       ``filter`` - filter number (within the specified filter group)


    *Using Filter Groups*

        Filter groups are filter criteria in the form of { field: value }

            Group 0 Filter 0                        ->      Filter 0
            Group 0 Filter 0 + Group 0 Filter 1     ->      Filter 0 OR Filter 1
            Group 0 Filter 0 + Group 1 Filter 0     ->      Filter 0 AND Filter 0
    """

    options = {
        'condition': condition,
        'group': self.last_group + 1,
        'filter': 0,
    }
    options.update(kwargs)

    criteria = (
            f'searchCriteria[filter_groups][{options["group"]}][filters][{options["filter"]}][field]={field}' +
            f'&searchCriteria[filter_groups][{options["group"]}][filters][{options["filter"]}][value]={value}' +
            f'&searchCriteria[filter_groups][{options["group"]}][filters][{options["filter"]}][condition_type]={options["condition"]}'
    )
    if not self.query.endswith('?'):
        self.query += '&'
    self.query += criteria
    return self

Add criteria to the search query

Parameters

• field – the API response field to search by

• value – the value of the field to compare

• condition – the comparison condition

• kwargs – additional search option arguments (group and filter)

Returns

the calling SearchQuery object

Return type

Self

Keyword Argument Options: Condition

The condition argument specifies the condition used to evaluate the attribute value

• "eq" (default): matches items for which field=value

• "gt": matches items for which field>value

• "lt": matches items for which field<value

• "gteq": matches items for which field>=value

• "lteq": matches items for which field<=value

• "in": matches items for which field in value.split(",")

  • Tip: for in, use by_list() if not building a complex query

Example

# Search for Orders created in 2023
>>> orders = api.orders.add_criteria(
...     field="created_at",
...     value="2023-01-01",
...     condition='gteq'
... ).execute()

Keyword Argument Options: Using Filter Groups

group - filter group number

filter - filter number (within the specified filter group)

Using Filter Groups

Filter groups are filter criteria in the form of { field: value }

Group 0 Filter 0 -> Filter 0 Group 0 Filter 0 + Group 0 Filter 1 -> Filter 0 OR Filter 1 Group 0 Filter 0 + Group 1 Filter 0 -> Filter 0 AND Filter 0

restrict_fields(fields)View on GitHub

View Source Code

def restrict_fields(self, fields: Iterable[str]) -> Self:
    """Constrain the API response data to only contain the specified fields

    :param fields: an iterable or comma separated string of fields to include in the response
    :returns: the calling SearchQuery object
    """
    if not isinstance(fields, str):
        if not isinstance(fields, Iterable):
            raise TypeError('"fields" must be a comma separated string or an iterable')
        fields = ','.join(fields)

    if (id_field := self.Model.IDENTIFIER) not in fields:
        fields += f',{id_field}'

    self.fields = f'&fields=items[{fields}]'
    return self

Constrain the API response data to only contain the specified fields

Parameters

fields (Iterable[str]) – an iterable or comma separated string of fields to include in the response

Returns

the calling SearchQuery object

Return type

Self

execute()View on GitHub

View Source Code

def execute(self) -> Optional[Model | List[Model]]:
    """Sends the search request using the current :attr:`~.scope` of the :attr:`client`

    .. tip:: Change the :attr:`.Client.scope` to retrieve :attr:`~.result` data
       from different store :attr:`~.views`

    :returns: the search query :attr:`~.result`
    """
    response = self.client.get(self.query + self.fields)
    self.__dict__.pop('result', None)
    self._result = response.json()
    return self.result

Sends the search request using the current scope of the client

Tip

Change the Client.scope to retrieve result data from different store views

Returns

the search query result

Return type

Optional[Union[Model, List[Model]]]

by_id(item_id)View on GitHub

View Source Code

def by_id(self, item_id: Union[int, str]) -> Optional[Model]:
    """Retrieve data for an individual item by its id

    .. note:: The ``id`` field used is different depending on the endpoint being queried

       * Most endpoints use an ``entity_id`` or ``id``
       * The ``orders/items`` endpoint uses ``item_id``
       * The ``products`` endpoint uses ``product_id``,
         but can also be queried :meth:`~ProductSearch.by_sku`

       The :attr:`~.Model.IDENTIFIER` attribute of each :class:`~.Model` contains the appropriate field

    :param item_id: id of the item to retrieve
    """
    self.query = self.query.strip('?') + str(item_id)
    return self.execute()

Retrieve data for an individual item by its id

Note

The id field used is different depending on the endpoint being queried

• Most endpoints use an entity_id or id

• The orders/items endpoint uses item_id

• The products endpoint uses product_id, but can also be queried by_sku()

The IDENTIFIER attribute of each Model contains the appropriate field

Parameters

item_id (Union[int, str]) – id of the item to retrieve

Return type

Optional[Model]

by_list(field, values)View on GitHub

View Source Code

def by_list(self, field: str, values: Iterable) -> Optional[Model, List[Model]]:
    """Search for multiple items using an iterable or comma-separated string of field values

    .. admonition:: Examples
       :class: example

       Retrieve :class:`~.Product` with ids from 1 to 10::

        # Values can be a list/tuple/iterable
        >> api.products.by_list('entity_id', range(1,11))

       Search for :class:`~.Order` that are processing, pending, or completed::

        #  Values can be a comma-separated string
        >> api.orders.by_list('status', 'processing,pending,completed')


    :param field: the API response field to search for matches in
    :param values: an iterable or comma separated string of values
    """
    if not isinstance(values, Iterable):
        raise TypeError('`values` must be an iterable')
    if not isinstance(values, str):
        values = ','.join(f'{value}' for value in values)
    return self.add_criteria(
        field=field,
        value=values,
        condition='in'
    ).execute()

Search for multiple items using an iterable or comma-separated string of field values

Examples

Retrieve Product with ids from 1 to 10:

# Values can be a list/tuple/iterable
>> api.products.by_list('entity_id', range(1,11))

Search for Order that are processing, pending, or completed:

#  Values can be a comma-separated string
>> api.orders.by_list('status', 'processing,pending,completed')

Parameters

• field (str) – the API response field to search for matches in

• values (Iterable) – an iterable or comma separated string of values

Return type

Optional[Model, List[Model]]

get_all()View on GitHub

View Source Code

def get_all(self) -> Optional[Model, List[Model]]:
    """Retrieve all items for the given search endpoint.

    .. warning:: Not guaranteed to work with all endpoints.
    """
    return self.since().execute()

Retrieve all items for the given search endpoint.

Warning

Not guaranteed to work with all endpoints.

Return type

Optional[Model, List[Model]]

since(sinceDate=None)View on GitHub

View Source Code

def since(self, sinceDate: str = None) -> Self:
    """Retrieve items for which ``created_at >= sinceDate``

    **Example**::

        # Retrieve products created in 2023
        >> api.products.since('2023-01-01').execute()


    .. tip:: Calling with no arguments retrieves all items

       ::

        # Retrieve all products
        >> api.products.since().execute()

    :param sinceDate: the date for response data to start from
    :return: the calling :class:`~SearchQuery`
    """
    return self.add_criteria(
        field='created_at',
        value=sinceDate,
        condition='gteq',
        group=self.last_group + 1,
    )

Retrieve items for which created_at >= sinceDate

Example:

# Retrieve products created in 2023
>> api.products.since('2023-01-01').execute()

Tip

Calling with no arguments retrieves all items

# Retrieve all products
>> api.products.since().execute()

Parameters

sinceDate (str) – the date for response data to start from

Returns

the calling SearchQuery

Return type

Self

until(toDate)View on GitHub

View Source Code

def until(self, toDate: str) -> Self:
    """Retrieve items for which ``created_at <= toDate``

    :param toDate: the date for response data to end at (inclusive)
    :return: the calling :class:`~SearchQuery`
    """
    return self.add_criteria(
        field='created_at',
        value=toDate,
        condition='lteq',
        group=self.last_group + 1,
    )

Retrieve items for which created_at <= toDate

Parameters

toDate (str) – the date for response data to end at (inclusive)

Returns

the calling SearchQuery

Return type

Self

property result: Optional[Union[Model, List[Model]]]View on GitHub

View Source Code

@cached_property
def result(self) -> Optional[Model | List[Model]]:
    """The result of the search query, wrapped by the :class:`~.Model` corresponding to the endpoint

    :returns: the API response as either an individual or list of :class:`~.Model` objects
    """
    result = self.validate_result()
    if result is None:
        return result
    if isinstance(result, list):
        return [self.parse(item) for item in result]
    if isinstance(result, dict):
        return self.parse(result)

The result of the search query, wrapped by the Model corresponding to the endpoint

Returns

the API response as either an individual or list of Model objects

validate_result()View on GitHub

View Source Code

def validate_result(self) -> Optional[Dict | List[Dict]]:
    """Parses the response and returns the actual result data, regardless of search approach"""
    if not self._result:
        return None

    if isinstance(self._result, list):
        return self._result

    if self._result.get('message'):  # Error; logged by Client
        return None

    if len(self._result.keys()) > 3:  # Single item, retrieved by id
        return self._result

    if 'items' in self._result:  # All successful responses with search criteria have `items` key
        items = self._result['items']
        if items:  # Response can still be {'items': None} though
            if len(items) > 1:
                return items
            return items[0]
        else:
            self.client.logger.info(
                "No matching {} for this search query".format(self.endpoint)
            )
            return None
    # I have no idea what could've gone wrong, sorry :/
    msg = "Search failed with an unknown error.\nAPI Response: {}".format(self._result)
    raise MagentoError(self.client, msg)

Parses the response and returns the actual result data, regardless of search approach

Return type

Optional[Union[Dict, List[Dict]]]

parse(data)View on GitHub

View Source Code

def parse(self, data: dict) -> Model:
    """Parses the API response with the corresponding :class:`~.Model` object

    :param data: API response data of a single item
    """
    if self.Model is not APIResponse:
        return self.Model(data, self.client)
    return self.Model(data, self.client, self.endpoint)

Parses the API response with the corresponding Model object

Parameters

data (dict) – API response data of a single item

Return type

Model

reset()View on GitHub

View Source Code

def reset(self) -> None:
    """Resets the query and result, allowing the object to be reused"""
    self._result = {}
    self.fields = ''
    self.query = self.client.url_for(self.endpoint) + '/?'
    self.__dict__.pop('result', None)

Resets the query and result, allowing the object to be reused

property result_count: intView on GitHub

View Source Code

@property
def result_count(self) -> int:
    """Number of items that matched the search criteria"""
    if not self._result or not self.result:
        return 0
    if isinstance(self.result, Model):
        return 1
    return len(self.result)

Number of items that matched the search criteria

property result_type: TypeView on GitHub

View Source Code

@property
def result_type(self) -> Type:
    """The type of the result"""
    return type(self.result)

The type of the result

property last_group: intView on GitHub

View Source Code

@property
def last_group(self) -> int:
    """The most recent filter group on the query
    
    :returns: the most recent filter group, or ``-1`` if no criteria has been added
    """
    if self.query.endswith('?'):
        return -1
    return int(re.match(
        r'.*searchCriteria\[filter_groups]\[(\d)]',
        self.query
    ).groups()[-1])

The most recent filter group on the query

Returns

the most recent filter group, or -1 if no criteria has been added

class magento.search.OrderSearch(client)View on GitHub

View Source Code

class OrderSearch(SearchQuery):

    """:class:`SearchQuery` subclass for the ``orders`` endpoint"""

    def __init__(self, client: Client):
        """Initialize an :class:`OrderSearch`

        :param client: an initialized :class:`~.Client` object
        """
        super().__init__(
            endpoint='orders',
            client=client,
            model=Order
        )

    def by_number(self, order_number: Union[int, str]) -> Optional[Order]:
        """Retrieve an :class:`~.Order` by number

        :param order_number: the order number (``increment_id``)
        """
        return self.add_criteria(
            field='increment_id',
            value=order_number
        ).execute()

    def by_product(self, product: Product) -> Optional[Order | List[Order]]:
        """Search for all :class:`~.Order` s of a :class:`~.Product`

        :param product: the :class:`~.Product` to search for in orders
        """
        items = self.client.order_items.by_product(product)
        return self.from_items(items)

    def by_sku(self, sku: str) -> Optional[Order | List[Order]]:
        """Search for :class:`~.Order` by product sku

        .. note:: Like :meth:`.OrderItemSearch.by_sku`, the sku will need to be an exact
           match to the sku of a simple product, including a custom option if applicable

           * Use :meth:`~.OrderSearch.by_product` or :meth:`~.OrderSearch.by_product_id`
             to find orders containing any of the :attr:`~.option_skus` and/or all
             :attr:`~.children` of a configurable product

        :param sku: the exact product sku to search for in orders
        """
        items = self.client.order_items.by_sku(sku)
        return self.from_items(items)

    def by_product_id(self, product_id: Union[int, str]) -> Optional[Order | List[Order]]:
        """Search for :class:`~.Order` s by ``product_id``

        :param product_id: the ``id`` (``product_id``) of the product to search for in orders
        """
        items = self.client.order_items.by_product_id(product_id)
        return self.from_items(items)

    def by_category_id(self, category_id: Union[int, str], search_subcategories: bool = False) -> Optional[Order | List[Order]]:
        """Search for :class:`~.Order` s by ``category_id``

        :param category_id: id of the category to search for in orders
        :param search_subcategories: if ``True``, also searches for orders from :attr:`~.all_subcategories`
        :returns: any :class:`~.Order` containing a :class:`~.Product` in the corresponding :class:`~.Category`
        """
        items = self.client.order_items.by_category_id(category_id, search_subcategories)
        return self.from_items(items)

    def by_category(self, category: Category, search_subcategories: bool = False) -> Optional[Order | List[Order]]:
        """Search for :class:`~.Order` s that contain any of the category's :attr:`~.Category.products`

        :param category: the :class:`~.Category` to use in the search
        :param search_subcategories: if ``True``, also searches for orders from :attr:`~.all_subcategories`
        :returns: any :class:`~.Order` that contains a product in the provided category
        """
        items = self.client.order_items.by_category(category, search_subcategories)
        return self.from_items(items)

    def by_skulist(self, skulist: Union[str, Iterable[str]]) -> Optional[Order | List[Order]]:
        """Search for :class:`~.Order` s using a list or comma separated string of product SKUs

        :param skulist: an iterable or comma separated string of product SKUs
        """
        items = self.client.order_items.by_skulist(skulist)
        return self.from_items(items)

    def by_customer(self, customer: Customer) -> Optional[Order | List[Order]]:
        """Search for :class:`~.Order` s by :class:`~.Customer`

        :param customer: the :class:`~.Customer` to retrieve orders from
        """
        return self.by_customer_id(customer.uid)

    def by_customer_id(self, customer_id: Union[int, str]) -> Optional[Order | List[Order]]:
        """Search for :class:`~.Order` s by ``customer_id``

        :param customer_id: the ``id`` of the customer to retrieve orders for
        """
        return self.add_criteria(
            field='customer_id',
            value=str(customer_id)
        ).execute()

    def from_items(self, items: Optional[OrderItem | List[OrderItem]]) -> Optional[Order, List[Order]]:
        """Retrieve unique :class:`~.Order` objects from :class:`~.OrderItem` entries using a single request

        :param items: an individual/list of order items
        """
        if items is None:
            return
        if isinstance(items, list):
            order_ids = set(item.order_id for item in items)
            return self.by_list('entity_id', order_ids)
        else:
            return items.order  # Single OrderItem

Bases: SearchQuery

SearchQuery subclass for the orders endpoint

__init__(client)View on GitHub

View Source Code

def __init__(self, client: Client):
    """Initialize an :class:`OrderSearch`

    :param client: an initialized :class:`~.Client` object
    """
    super().__init__(
        endpoint='orders',
        client=client,
        model=Order
    )

Initialize an OrderSearch

Parameters

client (Client) – an initialized Client object

by_number(order_number)View on GitHub

View Source Code

def by_number(self, order_number: Union[int, str]) -> Optional[Order]:
    """Retrieve an :class:`~.Order` by number

    :param order_number: the order number (``increment_id``)
    """
    return self.add_criteria(
        field='increment_id',
        value=order_number
    ).execute()

Retrieve an Order by number

Parameters

order_number (Union[int, str]) – the order number (increment_id)

Return type

Optional[Order]

by_product(product)View on GitHub

View Source Code

def by_product(self, product: Product) -> Optional[Order | List[Order]]:
    """Search for all :class:`~.Order` s of a :class:`~.Product`

    :param product: the :class:`~.Product` to search for in orders
    """
    items = self.client.order_items.by_product(product)
    return self.from_items(items)

Search for all Order s of a Product

Parameters

product (Product) – the Product to search for in orders

Return type

Optional[Union[Order, List[Order]]]

by_sku(sku)View on GitHub

View Source Code

def by_sku(self, sku: str) -> Optional[Order | List[Order]]:
    """Search for :class:`~.Order` by product sku

    .. note:: Like :meth:`.OrderItemSearch.by_sku`, the sku will need to be an exact
       match to the sku of a simple product, including a custom option if applicable

       * Use :meth:`~.OrderSearch.by_product` or :meth:`~.OrderSearch.by_product_id`
         to find orders containing any of the :attr:`~.option_skus` and/or all
         :attr:`~.children` of a configurable product

    :param sku: the exact product sku to search for in orders
    """
    items = self.client.order_items.by_sku(sku)
    return self.from_items(items)

Search for Order by product sku

Note

Like OrderItemSearch.by_sku(), the sku will need to be an exact match to the sku of a simple product, including a custom option if applicable

• Use by_product() or by_product_id() to find orders containing any of the option_skus and/or all children of a configurable product

Parameters

sku (str) – the exact product sku to search for in orders

Return type

Optional[Union[Order, List[Order]]]

by_product_id(product_id)View on GitHub

View Source Code

def by_product_id(self, product_id: Union[int, str]) -> Optional[Order | List[Order]]:
    """Search for :class:`~.Order` s by ``product_id``

    :param product_id: the ``id`` (``product_id``) of the product to search for in orders
    """
    items = self.client.order_items.by_product_id(product_id)
    return self.from_items(items)

Search for Order s by product_id

Parameters

product_id (Union[int, str]) – the id (product_id) of the product to search for in orders

Return type

Optional[Union[Order, List[Order]]]

by_category_id(category_id, search_subcategories=False)View on GitHub

View Source Code

def by_category_id(self, category_id: Union[int, str], search_subcategories: bool = False) -> Optional[Order | List[Order]]:
    """Search for :class:`~.Order` s by ``category_id``

    :param category_id: id of the category to search for in orders
    :param search_subcategories: if ``True``, also searches for orders from :attr:`~.all_subcategories`
    :returns: any :class:`~.Order` containing a :class:`~.Product` in the corresponding :class:`~.Category`
    """
    items = self.client.order_items.by_category_id(category_id, search_subcategories)
    return self.from_items(items)

Search for Order s by category_id

Parameters

• category_id (Union[int, str]) – id of the category to search for in orders

• search_subcategories (bool) – if True, also searches for orders from all_subcategories

Returns

any Order containing a Product in the corresponding Category

Return type

Optional[Union[Order, List[Order]]]

by_category(category, search_subcategories=False)View on GitHub

View Source Code

def by_category(self, category: Category, search_subcategories: bool = False) -> Optional[Order | List[Order]]:
    """Search for :class:`~.Order` s that contain any of the category's :attr:`~.Category.products`

    :param category: the :class:`~.Category` to use in the search
    :param search_subcategories: if ``True``, also searches for orders from :attr:`~.all_subcategories`
    :returns: any :class:`~.Order` that contains a product in the provided category
    """
    items = self.client.order_items.by_category(category, search_subcategories)
    return self.from_items(items)

Search for Order s that contain any of the category’s products

Parameters

• category (Category) – the Category to use in the search

• search_subcategories (bool) – if True, also searches for orders from all_subcategories

Returns

any Order that contains a product in the provided category

Return type

Optional[Union[Order, List[Order]]]

by_skulist(skulist)View on GitHub

View Source Code

def by_skulist(self, skulist: Union[str, Iterable[str]]) -> Optional[Order | List[Order]]:
    """Search for :class:`~.Order` s using a list or comma separated string of product SKUs

    :param skulist: an iterable or comma separated string of product SKUs
    """
    items = self.client.order_items.by_skulist(skulist)
    return self.from_items(items)

Search for Order s using a list or comma separated string of product SKUs

Parameters

skulist (Union[str, Iterable[str]]) – an iterable or comma separated string of product SKUs

Return type

Optional[Union[Order, List[Order]]]

by_customer(customer)View on GitHub

View Source Code

def by_customer(self, customer: Customer) -> Optional[Order | List[Order]]:
    """Search for :class:`~.Order` s by :class:`~.Customer`

    :param customer: the :class:`~.Customer` to retrieve orders from
    """
    return self.by_customer_id(customer.uid)

Search for Order s by Customer

Parameters

customer (Customer) – the Customer to retrieve orders from

Return type

Optional[Union[Order, List[Order]]]

by_customer_id(customer_id)View on GitHub

View Source Code

def by_customer_id(self, customer_id: Union[int, str]) -> Optional[Order | List[Order]]:
    """Search for :class:`~.Order` s by ``customer_id``

    :param customer_id: the ``id`` of the customer to retrieve orders for
    """
    return self.add_criteria(
        field='customer_id',
        value=str(customer_id)
    ).execute()

Search for Order s by customer_id

Parameters

customer_id (Union[int, str]) – the id of the customer to retrieve orders for

Return type

Optional[Union[Order, List[Order]]]

from_items(items)View on GitHub

View Source Code

def from_items(self, items: Optional[OrderItem | List[OrderItem]]) -> Optional[Order, List[Order]]:
    """Retrieve unique :class:`~.Order` objects from :class:`~.OrderItem` entries using a single request

    :param items: an individual/list of order items
    """
    if items is None:
        return
    if isinstance(items, list):
        order_ids = set(item.order_id for item in items)
        return self.by_list('entity_id', order_ids)
    else:
        return items.order  # Single OrderItem

Retrieve unique Order objects from OrderItem entries using a single request

Parameters

items (Optional[OrderItem | List[OrderItem]]) – an individual/list of order items

Return type

Optional[Order, List[Order]]

class magento.search.OrderItemSearch(client)View on GitHub

View Source Code

class OrderItemSearch(SearchQuery):

    """:class:`SearchQuery` subclass for the ``orders/items`` endpoint"""

    def __init__(self, client: Client):
        """Initialize an :class:`OrderItemSearch`

        :param client: an initialized :class:`~.Client` object
        """
        super().__init__(
            endpoint='orders/items',
            client=client,
            model=OrderItem
        )

    @cached_property
    def result(self) -> Optional[OrderItem | List[OrderItem]]:
        if result := super().result:
            if isinstance(result, list):
                return [item for item in result if item]
        return result

    def parse(self, data) -> Optional[OrderItem]:
        """Overrides :meth:`SearchQuery.parse` to fully hydrate :class:`~.OrderItem` objects

        Extra validation is required for OrderItems, as duplicated and/or incomplete data is returned
        when the child of a configurable product is searched :meth:`by_sku` or :meth:`by_product`

        :param data: API response data
        """
        if data.get('parent_item'):
            return None
        if parent_id := data.get('parent_item_id'):
            return self.client.order_items.by_id(parent_id)
        else:
            return OrderItem(data, self.client)

    def by_product(self, product: Product) -> Optional[OrderItem | List[OrderItem]]:
        """Search for :class:`~.OrderItem` entries by :class:`~.Product`

        .. note:: This will match OrderItems that contain

           * Any of the child products of a configurable product
           * Any of the :attr:`~.option_skus` of a product with custom options

        :param product: the :class:`~.Product` to search for in order items
        """
        if not isinstance(product, Product):
            raise TypeError(f'`product` must be of type {Product}')

        if items := self.by_product_id(product.id):
            return items

        self.reset()
        return self.by_sku(product.encoded_sku)

    def by_sku(self, sku: str) -> Optional[OrderItem | List[OrderItem]]:
        """Search for :class:`~.OrderItem` entries by product sku.

        .. admonition:: The SKU must be an exact match to the OrderItem SKU

           OrderItems always use the SKU of a simple product, including any custom options.
           This means that:

           * Searching the SKU of a configurable product returns nothing
           * If a product has custom options, the search will only find OrderItems
             that contain the specific option sku (or base sku) that's provided

           To search for OrderItems containing all :attr:`~.children` of a
           configurable product and/or all possible :attr:`~.option_skus`,
           use :meth:`~.by_product` or :meth:`~.by_product_id`

        :param sku: the exact product sku to search for in order items
        """
        return self.add_criteria('sku', Model.encode(sku)).execute()

    def by_product_id(self, product_id: Union[int, str]) -> Optional[OrderItem | List[OrderItem]]:
        """Search for :class:`~.OrderItem` entries by product id.

        :param product_id: the ``id`` (``product_id``) of the :class:`~.Product` to search for in order items
        """
        return self.add_criteria('product_id', product_id).execute()

    def by_category_id(self, category_id: Union[int, str], search_subcategories: bool = False) -> Optional[OrderItem | List[OrderItem]]:
        """Search for :class:`~.OrderItem` entries by ``category_id``

        :param category_id: id of the :class:`~.Category` to search for in order items
        :param search_subcategories: if ``True``, also searches for order items from :attr:`~.all_subcategories`
        :returns: any :class:`~.OrderItem` containing a :class:`~.Product` in the corresponding :class:`~.Category`
        """
        if category := self.client.categories.by_id(category_id):
            return self.by_category(category, search_subcategories)

    def by_category(self, category: Category, search_subcategories: bool = False) -> Optional[OrderItem | List[OrderItem]]:
        """Search for :class:`~.OrderItem` entries that contain any of the category's :attr:`~.Category.products`

        :param category: the :class:`~.Category` to use in the search
        :param search_subcategories: if ``True``, also searches for order items from :attr:`~.all_subcategories`
        """
        if not isinstance(category, Category):
            raise TypeError(f'`category` must be of type {Category}')

        product_ids = category.all_product_ids if search_subcategories else category.product_ids
        return self.by_list('product_id', product_ids)

    def by_skulist(self, skulist: Union[str, Iterable[str]]) -> Optional[OrderItem | List[OrderItem]]:
        """Search for :class:`~.OrderItem`s using a list or comma-separated string of product SKUs

        :param skulist: an iterable or comma separated string of product SKUs
        """
        if not isinstance(skulist, Iterable):
            raise TypeError(f'`skulist` must be an iterable or comma-separated string of SKUs')
        if isinstance(skulist, str):
            skulist = skulist.split(',')

        skus = map(Model.encode, skulist)
        return self.by_list('sku', skus)

Bases: SearchQuery

SearchQuery subclass for the orders/items endpoint

__init__(client)View on GitHub

View Source Code

def __init__(self, client: Client):
    """Initialize an :class:`OrderItemSearch`

    :param client: an initialized :class:`~.Client` object
    """
    super().__init__(
        endpoint='orders/items',
        client=client,
        model=OrderItem
    )

Initialize an OrderItemSearch

Parameters

client (Client) – an initialized Client object

property result: Optional[Union[OrderItem, List[OrderItem]]]View on GitHub

View Source Code

@cached_property
def result(self) -> Optional[OrderItem | List[OrderItem]]:
    if result := super().result:
        if isinstance(result, list):
            return [item for item in result if item]
    return result

The result of the search query, wrapped by the Model corresponding to the endpoint

Returns

the API response as either an individual or list of Model objects

parse(data)View on GitHub

View Source Code

def parse(self, data) -> Optional[OrderItem]:
    """Overrides :meth:`SearchQuery.parse` to fully hydrate :class:`~.OrderItem` objects

    Extra validation is required for OrderItems, as duplicated and/or incomplete data is returned
    when the child of a configurable product is searched :meth:`by_sku` or :meth:`by_product`

    :param data: API response data
    """
    if data.get('parent_item'):
        return None
    if parent_id := data.get('parent_item_id'):
        return self.client.order_items.by_id(parent_id)
    else:
        return OrderItem(data, self.client)

Overrides SearchQuery.parse() to fully hydrate OrderItem objects

Extra validation is required for OrderItems, as duplicated and/or incomplete data is returned when the child of a configurable product is searched by_sku() or by_product()

Parameters

data – API response data

Return type

Optional[OrderItem]

by_product(product)View on GitHub

View Source Code

def by_product(self, product: Product) -> Optional[OrderItem | List[OrderItem]]:
    """Search for :class:`~.OrderItem` entries by :class:`~.Product`

    .. note:: This will match OrderItems that contain

       * Any of the child products of a configurable product
       * Any of the :attr:`~.option_skus` of a product with custom options

    :param product: the :class:`~.Product` to search for in order items
    """
    if not isinstance(product, Product):
        raise TypeError(f'`product` must be of type {Product}')

    if items := self.by_product_id(product.id):
        return items

    self.reset()
    return self.by_sku(product.encoded_sku)

Search for OrderItem entries by Product

Note

This will match OrderItems that contain

• Any of the child products of a configurable product

• Any of the option_skus of a product with custom options

Parameters

product (Product) – the Product to search for in order items

Return type

Optional[Union[OrderItem, List[OrderItem]]]

by_sku(sku)View on GitHub

View Source Code

def by_sku(self, sku: str) -> Optional[OrderItem | List[OrderItem]]:
    """Search for :class:`~.OrderItem` entries by product sku.

    .. admonition:: The SKU must be an exact match to the OrderItem SKU

       OrderItems always use the SKU of a simple product, including any custom options.
       This means that:

       * Searching the SKU of a configurable product returns nothing
       * If a product has custom options, the search will only find OrderItems
         that contain the specific option sku (or base sku) that's provided

       To search for OrderItems containing all :attr:`~.children` of a
       configurable product and/or all possible :attr:`~.option_skus`,
       use :meth:`~.by_product` or :meth:`~.by_product_id`

    :param sku: the exact product sku to search for in order items
    """
    return self.add_criteria('sku', Model.encode(sku)).execute()

Search for OrderItem entries by product sku.

The SKU must be an exact match to the OrderItem SKU

OrderItems always use the SKU of a simple product, including any custom options. This means that:

• Searching the SKU of a configurable product returns nothing

• If a product has custom options, the search will only find OrderItems that contain the specific option sku (or base sku) that’s provided

To search for OrderItems containing all children of a configurable product and/or all possible option_skus, use by_product() or by_product_id()

Parameters

sku (str) – the exact product sku to search for in order items

Return type

Optional[Union[OrderItem, List[OrderItem]]]

by_product_id(product_id)View on GitHub

View Source Code

def by_product_id(self, product_id: Union[int, str]) -> Optional[OrderItem | List[OrderItem]]:
    """Search for :class:`~.OrderItem` entries by product id.

    :param product_id: the ``id`` (``product_id``) of the :class:`~.Product` to search for in order items
    """
    return self.add_criteria('product_id', product_id).execute()

Search for OrderItem entries by product id.

Parameters

product_id (Union[int, str]) – the id (product_id) of the Product to search for in order items

Return type

Optional[Union[OrderItem, List[OrderItem]]]

by_category_id(category_id, search_subcategories=False)View on GitHub

View Source Code

def by_category_id(self, category_id: Union[int, str], search_subcategories: bool = False) -> Optional[OrderItem | List[OrderItem]]:
    """Search for :class:`~.OrderItem` entries by ``category_id``

    :param category_id: id of the :class:`~.Category` to search for in order items
    :param search_subcategories: if ``True``, also searches for order items from :attr:`~.all_subcategories`
    :returns: any :class:`~.OrderItem` containing a :class:`~.Product` in the corresponding :class:`~.Category`
    """
    if category := self.client.categories.by_id(category_id):
        return self.by_category(category, search_subcategories)

Search for OrderItem entries by category_id

Parameters

• category_id (Union[int, str]) – id of the Category to search for in order items

• search_subcategories (bool) – if True, also searches for order items from all_subcategories

Returns

any OrderItem containing a Product in the corresponding Category

Return type

Optional[Union[OrderItem, List[OrderItem]]]

by_category(category, search_subcategories=False)View on GitHub

View Source Code

def by_category(self, category: Category, search_subcategories: bool = False) -> Optional[OrderItem | List[OrderItem]]:
    """Search for :class:`~.OrderItem` entries that contain any of the category's :attr:`~.Category.products`

    :param category: the :class:`~.Category` to use in the search
    :param search_subcategories: if ``True``, also searches for order items from :attr:`~.all_subcategories`
    """
    if not isinstance(category, Category):
        raise TypeError(f'`category` must be of type {Category}')

    product_ids = category.all_product_ids if search_subcategories else category.product_ids
    return self.by_list('product_id', product_ids)

Search for OrderItem entries that contain any of the category’s products

Parameters

• category (Category) – the Category to use in the search

• search_subcategories (bool) – if True, also searches for order items from all_subcategories

Return type

Optional[Union[OrderItem, List[OrderItem]]]

by_skulist(skulist)View on GitHub

View Source Code

def by_skulist(self, skulist: Union[str, Iterable[str]]) -> Optional[OrderItem | List[OrderItem]]:
    """Search for :class:`~.OrderItem`s using a list or comma-separated string of product SKUs

    :param skulist: an iterable or comma separated string of product SKUs
    """
    if not isinstance(skulist, Iterable):
        raise TypeError(f'`skulist` must be an iterable or comma-separated string of SKUs')
    if isinstance(skulist, str):
        skulist = skulist.split(',')

    skus = map(Model.encode, skulist)
    return self.by_list('sku', skus)

Search for :class:`~.OrderItem`s using a list or comma-separated string of product SKUs

Parameters

skulist (Union[str, Iterable[str]]) – an iterable or comma separated string of product SKUs

Return type

Optional[Union[OrderItem, List[OrderItem]]]

class magento.search.InvoiceSearch(client)View on GitHub

View Source Code

class InvoiceSearch(SearchQuery):

    """:class:`SearchQuery` subclass for the ``invoices`` endpoint"""

    def __init__(self, client: Client):
        """Initialize an :class:`InvoiceSearch`

        :param client: an initialized :class:`~.Client` object
        """
        super().__init__(
            endpoint='invoices',
            client=client,
            model=Invoice
        )

    def by_number(self, invoice_number: Union[int, str]) -> Optional[Invoice]:
        """Retrieve an :class:`~.Invoice` by number

        :param invoice_number: the invoice number (``increment_id``)
        """
        return self.add_criteria(
            field='increment_id',
            value=invoice_number
        ).execute()

    def by_order_number(self, order_number: Union[int, str]) -> Optional[Invoice]:
        """Retrieve an :class:`~.Invoice` by order number

        :param order_number: the order number (``increment_id``)
        """
        if order := self.client.orders.by_number(order_number):
            return self.by_order(order)

    def by_order(self, order: Order) -> Optional[Invoice]:
        """Retrieve the :class:`~.Invoice` for an :class:`~.Order`

        :param order: the :class:`~.Order` object to retrieve an invoice for
        """
        return self.by_order_id(order.id)

    def by_order_id(self, order_id: Union[int, str]) -> Optional[Invoice]:
        """Retrieve an :class:`~.Invoice` by ``order_id``

        :param order_id: the ``order_id`` of the order to retrieve an invoice for
        """
        return self.add_criteria(
            field='order_id',
            value=order_id
        ).execute()

    def by_product(self, product: Product) -> Optional[Invoice | List[Invoice]]:
        """Search for all :class:`~.Invoice` s of a :class:`~.Product`

        :param product: the :class:`~.Product` to search for in invoices
        """
        items = self.client.order_items.by_product(product)
        return self.from_order_items(items)

    def by_sku(self, sku: str) -> Optional[Invoice | List[Invoice]]:
        """Search for :class:`~.Invoice` s by product sku

        .. note:: Like :meth:`.OrderItemSearch.by_sku`, the sku will need to be an exact
           match to the sku of a simple product, including a custom option if applicable

           * Use :meth:`~.InvoiceSearch.by_product` or :meth:`~.InvoiceSearch.by_product_id`
             to find orders containing any of the :attr:`~.option_skus` and/or all
             :attr:`~.children` of a configurable product

        :param sku: the exact product sku to search for in invoices
        """
        items = self.client.order_items.by_sku(sku)
        return self.from_order_items(items)

    def by_product_id(self, product_id: Union[int, str]) -> Optional[Invoice | List[Invoice]]:
        """Search for :class:`~.Invoice` s by ``product_id``

        :param product_id: the ``id`` (``product_id``) of the product to search for in invoices
        """
        items = self.client.order_items.by_product_id(product_id)
        return self.from_order_items(items)

    def by_category_id(self, category_id: Union[int, str], search_subcategories: bool = False) -> Optional[Invoice | List[Invoice]]:
        """Search for :class:`~.Invoice` s by ``category_id``

        :param category_id: id of the category to search for in orders
        :param search_subcategories: if ``True``, also searches for orders from :attr:`~.all_subcategories`
        :returns: any :class:`~.Invoice` containing a :class:`~.Product` in the corresponding :class:`~.Category`
        """
        items = self.client.order_items.by_category_id(category_id, search_subcategories)
        return self.from_order_items(items)

    def by_category(self, category: Category, search_subcategories: bool = False) -> Optional[Invoice | List[Invoice]]:
        """Search for :class:`~.Invoice` s that contain any of the category's :attr:`~.Category.products`

        :param category: the :class:`~.Category` to use in the search
        :param search_subcategories: if ``True``, also searches for orders from :attr:`~.all_subcategories`
        :returns: any :class:`~.Invoice` that contains a product in the provided category
        """
        items = self.client.order_items.by_category(category, search_subcategories)
        return self.from_order_items(items)

    def by_skulist(self, skulist: Union[str, Iterable[str]]) -> Optional[Invoice | List[Invoice]]:
        """Search for :class:`~.Invoice` s using a list or comma separated string of product SKUs

        :param skulist: an iterable or comma separated string of product SKUs
        """
        items = self.client.order_items.by_skulist(skulist)
        return self.from_order_items(items)

    def by_customer(self, customer: Customer) -> Optional[Invoice | List[Invoice]]:
        """Search for all :class:`~.Invoice` s of a :class:`~.Customer`

        :param customer: the :class:`~.Customer` to search for in invoices
        :returns: any :class:`~.Invoice` associated with the provided :class:`~.Customer`
        """
        return self.by_customer_id(customer.uid)

    def by_customer_id(self, customer_id: Union[int, str]) -> Optional[Invoice | List[Invoice]]:
        """Search for :class:`~.Invoice` s by ``customer_id``

        :param customer_id: the ``id`` of the customer to retrieve invoices for
        """
        orders = self.client.orders.by_customer_id(customer_id)

        if isinstance(orders, list):
            order_ids = set(order.id for order in orders)
            return self.by_list('order_id', order_ids)
        else:
            return self.by_order_id(orders.id)

    def from_order_items(self, items: Optional[OrderItem | List[OrderItem]]) -> Optional[Invoice, List[Invoice]]:
        """Retrieve unique :class:`~.Invoice` objects from :class:`~.OrderItem` entries using a single request

        .. tip:: Since there is no ``invoices/items`` endpoint, to search for invoices we must first do an
           :class:`OrderItemSearch`, then retrieve the ``order_ids`` and search :meth:`~.by_order_id`

        :param items: an individual/list of order items
        """
        if items is None:
            return self.client.logger.info(
                'No matching invoices for this search query'
            )
        if isinstance(items, list):
            order_ids = set(item.order_id for item in items)
            return self.by_list('order_id', order_ids)
        else:
            return self.by_order_id(items.order_id)  # Single OrderItem

Bases: SearchQuery

SearchQuery subclass for the invoices endpoint

__init__(client)View on GitHub

View Source Code

def __init__(self, client: Client):
    """Initialize an :class:`InvoiceSearch`

    :param client: an initialized :class:`~.Client` object
    """
    super().__init__(
        endpoint='invoices',
        client=client,
        model=Invoice
    )

Initialize an InvoiceSearch

Parameters

client (Client) – an initialized Client object

by_number(invoice_number)View on GitHub

View Source Code

def by_number(self, invoice_number: Union[int, str]) -> Optional[Invoice]:
    """Retrieve an :class:`~.Invoice` by number

    :param invoice_number: the invoice number (``increment_id``)
    """
    return self.add_criteria(
        field='increment_id',
        value=invoice_number
    ).execute()

Retrieve an Invoice by number

Parameters

invoice_number (Union[int, str]) – the invoice number (increment_id)

Return type

Optional[Invoice]

by_order_number(order_number)View on GitHub

View Source Code

def by_order_number(self, order_number: Union[int, str]) -> Optional[Invoice]:
    """Retrieve an :class:`~.Invoice` by order number

    :param order_number: the order number (``increment_id``)
    """
    if order := self.client.orders.by_number(order_number):
        return self.by_order(order)

Retrieve an Invoice by order number

Parameters

order_number (Union[int, str]) – the order number (increment_id)

Return type

Optional[Invoice]

by_order(order)View on GitHub

View Source Code

def by_order(self, order: Order) -> Optional[Invoice]:
    """Retrieve the :class:`~.Invoice` for an :class:`~.Order`

    :param order: the :class:`~.Order` object to retrieve an invoice for
    """
    return self.by_order_id(order.id)

Retrieve the Invoice for an Order

Parameters

order (Order) – the Order object to retrieve an invoice for

Return type

Optional[Invoice]

by_order_id(order_id)View on GitHub

View Source Code

def by_order_id(self, order_id: Union[int, str]) -> Optional[Invoice]:
    """Retrieve an :class:`~.Invoice` by ``order_id``

    :param order_id: the ``order_id`` of the order to retrieve an invoice for
    """
    return self.add_criteria(
        field='order_id',
        value=order_id
    ).execute()

Retrieve an Invoice by order_id

Parameters

order_id (Union[int, str]) – the order_id of the order to retrieve an invoice for

Return type

Optional[Invoice]

by_product(product)View on GitHub

View Source Code

def by_product(self, product: Product) -> Optional[Invoice | List[Invoice]]:
    """Search for all :class:`~.Invoice` s of a :class:`~.Product`

    :param product: the :class:`~.Product` to search for in invoices
    """
    items = self.client.order_items.by_product(product)
    return self.from_order_items(items)

Search for all Invoice s of a Product

Parameters

product (Product) – the Product to search for in invoices

Return type

Optional[Union[Invoice, List[Invoice]]]

by_sku(sku)View on GitHub

View Source Code

def by_sku(self, sku: str) -> Optional[Invoice | List[Invoice]]:
    """Search for :class:`~.Invoice` s by product sku

    .. note:: Like :meth:`.OrderItemSearch.by_sku`, the sku will need to be an exact
       match to the sku of a simple product, including a custom option if applicable

       * Use :meth:`~.InvoiceSearch.by_product` or :meth:`~.InvoiceSearch.by_product_id`
         to find orders containing any of the :attr:`~.option_skus` and/or all
         :attr:`~.children` of a configurable product

    :param sku: the exact product sku to search for in invoices
    """
    items = self.client.order_items.by_sku(sku)
    return self.from_order_items(items)

Search for Invoice s by product sku

Note

Like OrderItemSearch.by_sku(), the sku will need to be an exact match to the sku of a simple product, including a custom option if applicable

• Use by_product() or by_product_id() to find orders containing any of the option_skus and/or all children of a configurable product

Parameters

sku (str) – the exact product sku to search for in invoices

Return type

Optional[Union[Invoice, List[Invoice]]]

by_product_id(product_id)View on GitHub

View Source Code

def by_product_id(self, product_id: Union[int, str]) -> Optional[Invoice | List[Invoice]]:
    """Search for :class:`~.Invoice` s by ``product_id``

    :param product_id: the ``id`` (``product_id``) of the product to search for in invoices
    """
    items = self.client.order_items.by_product_id(product_id)
    return self.from_order_items(items)

Search for Invoice s by product_id

Parameters

product_id (Union[int, str]) – the id (product_id) of the product to search for in invoices

Return type

Optional[Union[Invoice, List[Invoice]]]

by_category_id(category_id, search_subcategories=False)View on GitHub

View Source Code

def by_category_id(self, category_id: Union[int, str], search_subcategories: bool = False) -> Optional[Invoice | List[Invoice]]:
    """Search for :class:`~.Invoice` s by ``category_id``

    :param category_id: id of the category to search for in orders
    :param search_subcategories: if ``True``, also searches for orders from :attr:`~.all_subcategories`
    :returns: any :class:`~.Invoice` containing a :class:`~.Product` in the corresponding :class:`~.Category`
    """
    items = self.client.order_items.by_category_id(category_id, search_subcategories)
    return self.from_order_items(items)

Search for Invoice s by category_id

Parameters

• category_id (Union[int, str]) – id of the category to search for in orders

• search_subcategories (bool) – if True, also searches for orders from all_subcategories

Returns

any Invoice containing a Product in the corresponding Category

Return type

Optional[Union[Invoice, List[Invoice]]]

by_category(category, search_subcategories=False)View on GitHub

View Source Code

def by_category(self, category: Category, search_subcategories: bool = False) -> Optional[Invoice | List[Invoice]]:
    """Search for :class:`~.Invoice` s that contain any of the category's :attr:`~.Category.products`

    :param category: the :class:`~.Category` to use in the search
    :param search_subcategories: if ``True``, also searches for orders from :attr:`~.all_subcategories`
    :returns: any :class:`~.Invoice` that contains a product in the provided category
    """
    items = self.client.order_items.by_category(category, search_subcategories)
    return self.from_order_items(items)

Search for Invoice s that contain any of the category’s products

Parameters

• category (Category) – the Category to use in the search

• search_subcategories (bool) – if True, also searches for orders from all_subcategories

Returns

any Invoice that contains a product in the provided category

Return type

Optional[Union[Invoice, List[Invoice]]]

by_skulist(skulist)View on GitHub

View Source Code

def by_skulist(self, skulist: Union[str, Iterable[str]]) -> Optional[Invoice | List[Invoice]]:
    """Search for :class:`~.Invoice` s using a list or comma separated string of product SKUs

    :param skulist: an iterable or comma separated string of product SKUs
    """
    items = self.client.order_items.by_skulist(skulist)
    return self.from_order_items(items)

Search for Invoice s using a list or comma separated string of product SKUs

Parameters

skulist (Union[str, Iterable[str]]) – an iterable or comma separated string of product SKUs

Return type

Optional[Union[Invoice, List[Invoice]]]

by_customer(customer)View on GitHub

View Source Code

def by_customer(self, customer: Customer) -> Optional[Invoice | List[Invoice]]:
    """Search for all :class:`~.Invoice` s of a :class:`~.Customer`

    :param customer: the :class:`~.Customer` to search for in invoices
    :returns: any :class:`~.Invoice` associated with the provided :class:`~.Customer`
    """
    return self.by_customer_id(customer.uid)

Search for all Invoice s of a Customer

Parameters

customer (Customer) – the Customer to search for in invoices

Returns

any Invoice associated with the provided Customer

Return type

Optional[Union[Invoice, List[Invoice]]]

by_customer_id(customer_id)View on GitHub

View Source Code

def by_customer_id(self, customer_id: Union[int, str]) -> Optional[Invoice | List[Invoice]]:
    """Search for :class:`~.Invoice` s by ``customer_id``

    :param customer_id: the ``id`` of the customer to retrieve invoices for
    """
    orders = self.client.orders.by_customer_id(customer_id)

    if isinstance(orders, list):
        order_ids = set(order.id for order in orders)
        return self.by_list('order_id', order_ids)
    else:
        return self.by_order_id(orders.id)

Search for Invoice s by customer_id

Parameters

customer_id (Union[int, str]) – the id of the customer to retrieve invoices for

Return type

Optional[Union[Invoice, List[Invoice]]]

from_order_items(items)View on GitHub

View Source Code

def from_order_items(self, items: Optional[OrderItem | List[OrderItem]]) -> Optional[Invoice, List[Invoice]]:
    """Retrieve unique :class:`~.Invoice` objects from :class:`~.OrderItem` entries using a single request

    .. tip:: Since there is no ``invoices/items`` endpoint, to search for invoices we must first do an
       :class:`OrderItemSearch`, then retrieve the ``order_ids`` and search :meth:`~.by_order_id`

    :param items: an individual/list of order items
    """
    if items is None:
        return self.client.logger.info(
            'No matching invoices for this search query'
        )
    if isinstance(items, list):
        order_ids = set(item.order_id for item in items)
        return self.by_list('order_id', order_ids)
    else:
        return self.by_order_id(items.order_id)  # Single OrderItem

Retrieve unique Invoice objects from OrderItem entries using a single request

Tip

Since there is no invoices/items endpoint, to search for invoices we must first do an OrderItemSearch, then retrieve the order_ids and search by_order_id()

Parameters

items (Optional[OrderItem | List[OrderItem]]) – an individual/list of order items

Return type

Optional[Invoice, List[Invoice]]

class magento.search.ProductSearch(client)View on GitHub

View Source Code

class ProductSearch(SearchQuery):

    """:class:`SearchQuery` subclass for the ``products`` endpoint"""

    def __init__(self, client: Client):
        """Initialize a :class:`ProductSearch`

        :param client: an initialized :class:`~.Client` object
        """
        super().__init__(
            endpoint='products',
            client=client,
            model=Product
        )

    @property
    def attributes(self) -> ProductAttributeSearch:
        """Alternate way to access the SearchQuery for :class:`~.ProductAttribute` data"""
        return ProductAttributeSearch(self.client)

    def by_id(self, item_id: Union[int, str]) -> Optional[Product]:
        """Retrieve a :class:`~.Product` by ``product_id``

        .. note:: Response data from the ``products`` endpoint only has an ``id`` field, but
           all other endpoints that return data about products will use ``product_id``

        :param item_id: the ``id`` (``product_id``) of the product
        """
        return self.add_criteria(
            field='entity_id',  # Product has no "entity_id" field in API responses
            value=item_id  # But to search by the "id" field, must use "entity_id"
        ).execute()

    def by_sku(self, sku) -> Optional[Product]:
        """Retrieve a :class:`~.Product` by ``sku``

        :param sku: the product sku
        """
        return super().by_id(Model.encode(sku))

    def by_skulist(self, skulist: Union[str, Iterable[str]]) -> Optional[Product | List[Product]]:
        """Search for :class:`~.Product`s using a list or comma separated string of SKUs

        :param skulist: an iterable or comma separated string of SKUs
        """
        if not isinstance(skulist, Iterable):
            raise TypeError(f'`skulist` must be an iterable or comma-separated string of SKUs')
        if isinstance(skulist, str):
            skulist = skulist.split(',')

        skus = map(Model.encode, skulist)
        return self.by_list('sku', skus)

    def by_category(self, category: Category, search_subcategories: bool = False) -> Optional[Product | List[Product]]:
        """Search for :class:`~.Product` s in a :class:`~.Category`

        :param category: the :class:`~.Category` to retrieve products from
        :param search_subcategories: if ``True``, also retrieves products from :attr:`~.all_subcategories`
        """
        if not isinstance(category, Category):
            raise TypeError(f'`category` must be of type {Category}')

        if search_subcategories:
            category_ids = [category.id] + category.all_subcategory_ids
            return self.by_list('category_id', category_ids)
        else:
            return self.add_criteria('category_id', category.id).execute()

    def by_category_id(self, category_id: Union[int, str], search_subcategories: bool = False) -> Optional[Product | List[Product]]:
        """Search for :class:`~.Product` s by ``category_id``

        :param category_id: the id of the :class:`~.Category` to retrieve products from
        :param search_subcategories: if ``True``, also retrieves products from :attr:`~.all_subcategories`
        """
        if search_subcategories:
            if category := self.client.categories.by_id(category_id):
                return self.by_category(category, search_subcategories)
            return None
        else:
            return self.add_criteria('category_id', category_id).execute()

    def by_customer_id(self, customer_id: Union[int, str], exclude_cancelled: bool = True):
        """Search for ordered :class:`~.Product`\s by ``customer_id``

        :param customer_id: the ``id`` of the customer to retrieve ordered products for
        :param exclude_cancelled: flag indicating if products from cancelled orders should be excluded
        :returns: products that the customer has ordered, as an individual or list of :class:`~.Product` objects
        """
        if customer := self.client.customers.by_id(customer_id):
            return customer.get_ordered_products(exclude_cancelled)

    def get_stock(self, sku) -> Optional[int]:
        """Retrieve the :attr:`~.stock` of a product by sku

        :param sku: the product sku
        """
        if product := self.by_sku(sku):
            return product.stock

Bases: SearchQuery

SearchQuery subclass for the products endpoint

__init__(client)View on GitHub

View Source Code

def __init__(self, client: Client):
    """Initialize a :class:`ProductSearch`

    :param client: an initialized :class:`~.Client` object
    """
    super().__init__(
        endpoint='products',
        client=client,
        model=Product
    )

Initialize a ProductSearch

Parameters

client (Client) – an initialized Client object

property attributes: ProductAttributeSearchView on GitHub

View Source Code

@property
def attributes(self) -> ProductAttributeSearch:
    """Alternate way to access the SearchQuery for :class:`~.ProductAttribute` data"""
    return ProductAttributeSearch(self.client)

Alternate way to access the SearchQuery for ProductAttribute data

by_id(item_id)View on GitHub

View Source Code

def by_id(self, item_id: Union[int, str]) -> Optional[Product]:
    """Retrieve a :class:`~.Product` by ``product_id``

    .. note:: Response data from the ``products`` endpoint only has an ``id`` field, but
       all other endpoints that return data about products will use ``product_id``

    :param item_id: the ``id`` (``product_id``) of the product
    """
    return self.add_criteria(
        field='entity_id',  # Product has no "entity_id" field in API responses
        value=item_id  # But to search by the "id" field, must use "entity_id"
    ).execute()

Retrieve a Product by product_id

Note

Response data from the products endpoint only has an id field, but all other endpoints that return data about products will use product_id

Parameters

item_id (Union[int, str]) – the id (product_id) of the product

Return type

Optional[Product]

by_sku(sku)View on GitHub

View Source Code

def by_sku(self, sku) -> Optional[Product]:
    """Retrieve a :class:`~.Product` by ``sku``

    :param sku: the product sku
    """
    return super().by_id(Model.encode(sku))

Retrieve a Product by sku

Parameters

sku – the product sku

Return type

Optional[Product]

by_skulist(skulist)View on GitHub

View Source Code

def by_skulist(self, skulist: Union[str, Iterable[str]]) -> Optional[Product | List[Product]]:
    """Search for :class:`~.Product`s using a list or comma separated string of SKUs

    :param skulist: an iterable or comma separated string of SKUs
    """
    if not isinstance(skulist, Iterable):
        raise TypeError(f'`skulist` must be an iterable or comma-separated string of SKUs')
    if isinstance(skulist, str):
        skulist = skulist.split(',')

    skus = map(Model.encode, skulist)
    return self.by_list('sku', skus)

Search for :class:`~.Product`s using a list or comma separated string of SKUs

Parameters

skulist (Union[str, Iterable[str]]) – an iterable or comma separated string of SKUs

Return type

Optional[Union[Product, List[Product]]]

by_category(category, search_subcategories=False)View on GitHub

View Source Code

def by_category(self, category: Category, search_subcategories: bool = False) -> Optional[Product | List[Product]]:
    """Search for :class:`~.Product` s in a :class:`~.Category`

    :param category: the :class:`~.Category` to retrieve products from
    :param search_subcategories: if ``True``, also retrieves products from :attr:`~.all_subcategories`
    """
    if not isinstance(category, Category):
        raise TypeError(f'`category` must be of type {Category}')

    if search_subcategories:
        category_ids = [category.id] + category.all_subcategory_ids
        return self.by_list('category_id', category_ids)
    else:
        return self.add_criteria('category_id', category.id).execute()

Search for Product s in a Category

Parameters

• category (Category) – the Category to retrieve products from

• search_subcategories (bool) – if True, also retrieves products from all_subcategories

Return type

Optional[Union[Product, List[Product]]]

by_category_id(category_id, search_subcategories=False)View on GitHub

View Source Code

def by_category_id(self, category_id: Union[int, str], search_subcategories: bool = False) -> Optional[Product | List[Product]]:
    """Search for :class:`~.Product` s by ``category_id``

    :param category_id: the id of the :class:`~.Category` to retrieve products from
    :param search_subcategories: if ``True``, also retrieves products from :attr:`~.all_subcategories`
    """
    if search_subcategories:
        if category := self.client.categories.by_id(category_id):
            return self.by_category(category, search_subcategories)
        return None
    else:
        return self.add_criteria('category_id', category_id).execute()

Search for Product s by category_id

Parameters

• category_id (Union[int, str]) – the id of the Category to retrieve products from

• search_subcategories (bool) – if True, also retrieves products from all_subcategories

Return type

Optional[Union[Product, List[Product]]]

by_customer_id(customer_id, exclude_cancelled=True)View on GitHub

View Source Code

def by_customer_id(self, customer_id: Union[int, str], exclude_cancelled: bool = True):
    """Search for ordered :class:`~.Product`\s by ``customer_id``

    :param customer_id: the ``id`` of the customer to retrieve ordered products for
    :param exclude_cancelled: flag indicating if products from cancelled orders should be excluded
    :returns: products that the customer has ordered, as an individual or list of :class:`~.Product` objects
    """
    if customer := self.client.customers.by_id(customer_id):
        return customer.get_ordered_products(exclude_cancelled)

Search for ordered Products by customer_id

Parameters

• customer_id (Union[int, str]) – the id of the customer to retrieve ordered products for

• exclude_cancelled (bool) – flag indicating if products from cancelled orders should be excluded

Returns

products that the customer has ordered, as an individual or list of Product objects

get_stock(sku)View on GitHub

View Source Code

def get_stock(self, sku) -> Optional[int]:
    """Retrieve the :attr:`~.stock` of a product by sku

    :param sku: the product sku
    """
    if product := self.by_sku(sku):
        return product.stock

Retrieve the stock of a product by sku

Parameters

sku – the product sku

Return type

Optional[int]

class magento.search.ProductAttributeSearch(client)View on GitHub

View Source Code

class ProductAttributeSearch(SearchQuery):

    """:class:`SearchQuery` subclass for the ``products/attributes`` endpoint"""

    def __init__(self, client: Client):
        """Initialize a :class:`ProductAttributeSearch`

        :param client: an initialized :class:`~.Client` object
        """
        super().__init__(
            endpoint='products/attributes',
            client=client,
            model=ProductAttribute
        )

    def get_all(self) -> Optional[List[ProductAttribute]]:
        """Retrieve a list of all :class:`~.ProductAttribute`s"""
        return self.add_criteria('position', 0, 'gteq').execute()

    def by_code(self, attribute_code: str) -> Optional[ProductAttribute]:
        """Retrieve a :class:`~.ProductAttribute` by its attribute code

        :param attribute_code: the code of the :class:`~.ProductAttribute`
        """
        return self.by_id(attribute_code)

    def get_types(self) -> Optional[List[APIResponse]]:
        """Retrieve a list of all available :class:`~.ProductAttribute` types"""
        return self.client.search(f'{self.endpoint}/types').execute()

Bases: SearchQuery

SearchQuery subclass for the products/attributes endpoint

__init__(client)View on GitHub

View Source Code

def __init__(self, client: Client):
    """Initialize a :class:`ProductAttributeSearch`

    :param client: an initialized :class:`~.Client` object
    """
    super().__init__(
        endpoint='products/attributes',
        client=client,
        model=ProductAttribute
    )

Initialize a ProductAttributeSearch

Parameters

client (Client) – an initialized Client object

get_all()View on GitHub

View Source Code

def get_all(self) -> Optional[List[ProductAttribute]]:
    """Retrieve a list of all :class:`~.ProductAttribute`s"""
    return self.add_criteria('position', 0, 'gteq').execute()

Retrieve a list of all :class:`~.ProductAttribute`s

Return type

Optional[List[ProductAttribute]]

by_code(attribute_code)View on GitHub

View Source Code

def by_code(self, attribute_code: str) -> Optional[ProductAttribute]:
    """Retrieve a :class:`~.ProductAttribute` by its attribute code

    :param attribute_code: the code of the :class:`~.ProductAttribute`
    """
    return self.by_id(attribute_code)

Retrieve a ProductAttribute by its attribute code

Parameters

attribute_code (str) – the code of the ProductAttribute

Return type

Optional[ProductAttribute]

get_types()View on GitHub

View Source Code

def get_types(self) -> Optional[List[APIResponse]]:
    """Retrieve a list of all available :class:`~.ProductAttribute` types"""
    return self.client.search(f'{self.endpoint}/types').execute()

Retrieve a list of all available ProductAttribute types

Return type

Optional[List[APIResponse]]

class magento.search.CustomerSearch(client)View on GitHub

View Source Code

class CustomerSearch(SearchQuery):
    """:class:`SearchQuery` subclass for the ``customers/search`` endpoint"""

    def __init__(self, client: Client):
        """Initialize a :class:`CustomerSearch`

        :param client: an initialized :class:`~.Client` object
        """
        super().__init__(
            endpoint='customers/search',
            client=client,
            model=Customer
        )

    def by_id(self, item_id: Union[int, str]) -> Optional[Customer]:
        self.query = self.query.replace('customers/search', 'customers')
        return super().by_id(item_id)

    def by_first_name(self, name):
        return self.add_criteria('firstName', name).execute()

    def by_last_name(self, name):
        return self.add_criteria('lastName', name).execute()

    def by_invoice(self, invoice: Invoice):
        return self.by_order(invoice.order)

    def by_order(self, order: Order):
        if customer_id := order.data.get("customer_id"):
            return self.by_id(customer_id)
        else:
            return self.client.logger.info(
                f"No customer account exists for {order}")

    def by_product(self, product: Product) -> Optional[Customer | List[Customer]]:
        orders = product.get_orders() or []
        customer_ids = set()

        if not isinstance(orders, list):
            return self.by_order(orders)

        for order in orders:
            if customer_id := order.data.get('customer_id'):
                customer_ids.add(customer_id)

        return self.by_list('entity_id', customer_ids)

Bases: SearchQuery

SearchQuery subclass for the customers/search endpoint

__init__(client)View on GitHub

View Source Code

def __init__(self, client: Client):
    """Initialize a :class:`CustomerSearch`

    :param client: an initialized :class:`~.Client` object
    """
    super().__init__(
        endpoint='customers/search',
        client=client,
        model=Customer
    )

Initialize a CustomerSearch

Parameters

client (Client) – an initialized Client object

by_id(item_id)View on GitHub

View Source Code

def by_id(self, item_id: Union[int, str]) -> Optional[Customer]:
    self.query = self.query.replace('customers/search', 'customers')
    return super().by_id(item_id)

Retrieve data for an individual item by its id

Note

The id field used is different depending on the endpoint being queried

• Most endpoints use an entity_id or id

• The orders/items endpoint uses item_id

• The products endpoint uses product_id, but can also be queried by_sku()

The IDENTIFIER attribute of each Model contains the appropriate field

Parameters

item_id (Union[int, str]) – id of the item to retrieve

Return type

Optional[Customer]

by_first_name(name)View on GitHub

View Source Code

def by_first_name(self, name):
    return self.add_criteria('firstName', name).execute()

by_last_name(name)View on GitHub

View Source Code

def by_last_name(self, name):
    return self.add_criteria('lastName', name).execute()

by_invoice(invoice)View on GitHub

View Source Code

def by_invoice(self, invoice: Invoice):
    return self.by_order(invoice.order)

by_order(order)View on GitHub

View Source Code

def by_order(self, order: Order):
    if customer_id := order.data.get("customer_id"):
        return self.by_id(customer_id)
    else:
        return self.client.logger.info(
            f"No customer account exists for {order}")

by_product(product)View on GitHub

View Source Code

def by_product(self, product: Product) -> Optional[Customer | List[Customer]]:
    orders = product.get_orders() or []
    customer_ids = set()

    if not isinstance(orders, list):
        return self.by_order(orders)

    for order in orders:
        if customer_id := order.data.get('customer_id'):
            customer_ids.add(customer_id)

    return self.by_list('entity_id', customer_ids)

Return type

Optional[Union[Customer, List[Customer]]]

class magento.search.CategorySearch(client)View on GitHub

View Source Code

class CategorySearch(SearchQuery):

    """:class:`SearchQuery` subclass for the ``categories`` endpoint"""

    def __init__(self, client: Client):
        """Initialize a :class:`CategorySearch`

        :param client: an initialized :class:`~.Client` object
        """
        super().__init__(
            endpoint='categories',
            client=client,
            model=Category
        )

    def by_id(self, item_id: Union[int, str]) -> Optional[Category]:
        self.query += f'rootCategoryId={item_id}'
        return self.execute()

    def by_list(self, field: str, values: Iterable) -> Optional[Category, List[Category]]:
        self.query = self.query.replace('categories', 'categories/list')
        return super().by_list(field, values)

    def get_root(self) -> Category:
        """Retrieve the top level/default :class:`~.Category` (every other category is a subcategory)"""
        return self.execute()

    def get_all(self) -> List[Category]:
        """Retrieve a list of all categories"""
        self.query = self.query.replace('categories', 'categories/list') + 'searchCriteria[currentPage]=1'
        return self.execute()

    def by_name(self, name: str, exact: bool = True) -> Optional[Category | List[Category]]:
        """Search for a :class:`~.Category` by name

        :param name: the category name to search for
        :param exact: whether the name should be an exact match
        """
        self.query = self.query.replace('categories', 'categories/list')
        if exact:
            return self.add_criteria('name', name).execute()
        else:
            return self.add_criteria('name', f'%25{name}%25', 'like').execute()

Bases: SearchQuery

SearchQuery subclass for the categories endpoint

__init__(client)View on GitHub

View Source Code

def __init__(self, client: Client):
    """Initialize a :class:`CategorySearch`

    :param client: an initialized :class:`~.Client` object
    """
    super().__init__(
        endpoint='categories',
        client=client,
        model=Category
    )

Initialize a CategorySearch

Parameters

client (Client) – an initialized Client object

get_root()View on GitHub

View Source Code

def get_root(self) -> Category:
    """Retrieve the top level/default :class:`~.Category` (every other category is a subcategory)"""
    return self.execute()

Retrieve the top level/default Category (every other category is a subcategory)

Return type

Category

get_all()View on GitHub

View Source Code

def get_all(self) -> List[Category]:
    """Retrieve a list of all categories"""
    self.query = self.query.replace('categories', 'categories/list') + 'searchCriteria[currentPage]=1'
    return self.execute()

Retrieve a list of all categories

Return type

List[Category]

by_name(name, exact=True)View on GitHub

View Source Code

def by_name(self, name: str, exact: bool = True) -> Optional[Category | List[Category]]:
    """Search for a :class:`~.Category` by name

    :param name: the category name to search for
    :param exact: whether the name should be an exact match
    """
    self.query = self.query.replace('categories', 'categories/list')
    if exact:
        return self.add_criteria('name', name).execute()
    else:
        return self.add_criteria('name', f'%25{name}%25', 'like').execute()

Search for a Category by name

Parameters

• name (str) – the category name to search for

• exact (bool) – whether the name should be an exact match

Return type

Optional[Union[Category, List[Category]]]