shopware-api-client 2.0.0

An api client for the Shopware API

Shopware API Client

A Django-ORM like, Python 3.12+, async Shopware 6 admin and store-front API client.

Installation

pip install shopware-api-client

Note: Redis support has been refactored in version 2.0. If you need custom caching backends like Redis, you can provide your own cache implementation through the config's cache parameter.

Usage

There are two kinds of clients provided by this library. The client.AdminClient for the Admin API and the client.StoreClient for the Store API.

client.AdminClient

To use the AdminClient you need to create a config.AdminConfig. The AdminConfig class supports two login methods (grant-types):

• client_credentials (Default) Let's you log in with a client_id and client_secret
• password Let's you log in using a username and password

You also need to provide the Base-URL of your shop.

Example:

from shopware_api_client.config import AdminConfig

CLIENT_ID = "MyClientID"
CLIENT_SECRET = "SuperSecretToken"
SHOP_URL = "https://pets24.shop"

config = AdminConfig(url=SHOP_URL, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, grant_type="client_credentials")

# or for "password"

ADMIN_USER = "admin"
ADMIN_PASSWORD = "!MeowMoewMoew~"

config = AdminConfig(url=SHOP_URL, username=ADMIN_USER, password=ADMIN_PASSWORD, grant_type="password")

Now you can create the Client:

from shopware_api_client.client import AdminClient

client = AdminClient(config=config)

Client-Connections should be closed after usage: await client.close(). The client can also be used in an async with block to be closed automatically.

from shopware_api_client.client import AdminClient

async with AdminClient(config=config) as client:
    # do important stuff
    pass

All registered Endpoints are directly available from the client instance. For example if you want to query the Customer Endpoint:

customer = await client.customer.first()

All available Endpoint functions can be found in the AdminEndpoint section.

You can also use the Endpoint-Class directly or the associated Pydantic Model:

from shopware_api_client.endpoints.admin.core.customer import Customer, CustomerEndpoint

# Endpoint
customer_endpoint = CustomerEndpoint(client=client)
customer = await customer_endpoint.first()

# Pydantic Model
customer = await Customer.using(client=client).first()

Related Objects

You can use returned Pydantic model objects to access their related objects:

from shopware_api_client.endpoints.admin import Customer

customer = await Customer.using(client=client).first()
customer_group = await customer.group  # Returns a CustomerGroup object
all_the_customers = await customer_group.customers  # Returns a list of Customer objects

!! Be careful to not close the client before doing related objects calls, since they use the same Client instance !!

from shopware_api_client.client import AdminClient
from shopware_api_client.endpoints.admin import Customer

async with AdminClient(config=config) as client:
    customer = await Customer.using(client=client).first()

customer_group = await customer.group  # This will fail, because the client connection is already closed!

CustomEntities

Shopware allows to create custom entities. You can use the load_custom_entities function to load them into the client.

from shopware_api_client.client import AdminClient

config = ...
client = AdminClient(config=config)
await client.load_custom_entities(client)

# Endpoint for the custom entity ce_blog
await client.ce_blog.all()

# Pydantic Model for the custom entity ce_blog
CeBlog = client.ce_blog.model_class

Since custom entities are completely dynamic no autocompletion in IDE is available. However there are some pydantic validations added for the field-types of the custom entity. Relations are currently not supported, but everything else should work as expected.

client.StoreClient

To use the StoreClient you need to create a config.StoreConfig. The StoreConfig needs a store api access key. You also need to provide the Base-URL of your shop.

Some Endpoints (that are somehow related to a user) require a context-token. This parameter is optional.

Example:

from shopware_api_client.config import StoreConfig

ACCESS_KEY = "SJMSAKSOMEKEY"
CONTEXT_TOKEN = "ASKSKJNNMMS"
SHOP_URL = "https://pets24.shop"

config = StoreConfig(url=SHOP_URL, access_key=STORE_API_ACCESS_KEY, context_token=CONTEXT_TOKEN)

This config can be used with the StoreClient, which works exactly like the AdminClient. The StoreClient has far less endpoints and does mostly not support full updated of models, but uses helper-functions.

Caching and Rate Limits

By default, both clients use an in-memory DictCache for caching. If you need to share caches across multiple client instances (e.g., running in different processes), you can provide a custom cache implementation by passing a cache parameter to the config.

The cache parameter accepts any object implementing the CacheProtocol:

class CacheProtocol(Protocol):
    async def get(self, key: str) -> Any | None: ...
    async def set(self, key: str, value: Any, ttl: int | None = None) -> None: ...
    async def delete(self, key: str) -> None: ...

Example: Using a Custom Redis Cache

from shopware_api_client.config import AdminConfig, StoreConfig
from shopware_api_client.client import AdminClient, StoreClient

# Implement a Redis-based cache using your preferred Redis client
class RedisCache:
    def __init__(self, redis_client):
        self.client = redis_client

    async def get(self, key: str):
        value = await self.client.get(key)
        return json.loads(value) if value else None

    async def set(self, key: str, value: Any, ttl: int | None = None):
        await self.client.set(key, json.dumps(value), ex=ttl)

    async def delete(self, key: str):
        await self.client.delete(key)

# Use the custom cache with your config
redis_cache = RedisCache(your_redis_client)

admin_config = AdminConfig(
    url='https://your-shop.com',
    client_id='...',
    client_secret='...',
    cache=redis_cache,  # Pass the custom cache
)
admin_client = AdminClient(config=admin_config)

store_config = StoreConfig(
    url='https://your-shop.com',
    access_key='...',
    cache=redis_cache,  # Reuse the same cache across clients
)
store_client = StoreClient(config=store_config)

AdminEndpoint

The base.AdminEndpoint class should be used for creating new Admin-Endpoints. It provides some usefull functions to call the Shopware-API.

The base structure of an Endpoint is pretty simple:

from shopware_api_client.base import EndpointMixin, AdminEndpoint


class CustomerGroup(EndpointMixin["CustomerGroupEndpoint"]):
    # Model definition
    pass


class CustomerGroupEndpoint(AdminEndpoint[CustomerGroup]):
    name = "customer_group"  # name of the Shopware-Endpoint (snaky)
    path = "/customer-group"  # path of the Shopware-Endpoint
    model_class = CustomerGroup  # Pydantic-Model of this Endpoint

List of available Functions

• all() return all objects (GET /customer-group or POST /search/customer-group if filter or sort is set)
• get(pk: str = id) returns the object with the passed id (GET /customer-group/id)
• update(pk: str = id, obj: ModelClass | dict[str: Any]) updates an object (PATCH /customer-group/id)
• create(obj: ModelClass | dict[str: Any]) creates a new object (POST /customer-group)
• delete(pk: str = id) deletes an object (DELETE /customer-group/id)
• filter(name="Cats") adds a filter to the query. Needs to be called with .all(), .iter() or .first())) More Info: Filter
• limit(count: int | None) sets the limit parameter, to limit the amount of results. Needs to be called with .all() or .first()
• first() sets the limit to 1 and returns the result (calling .all())
• order_by(fields: str | tuple[str] sets the sort parameter. Needs to be called with .all(), .iter() or .first(). Syntax: "name" for ASC, "-name" for DESC
• select_related(**kwargs: dict[str, Any]) sets the _associations parameter to define which related models to load in the request. Needs to be called with .all(), .iter() or .first().
• only(**kwargs: list[str]) sets the _includes parameter to define which fields to request. Needs to be called with .all(), .iter() or .first().
• iter(batch_size: int = 100) sets the limit-parameter to batch_size and makes use of the pagination of the api. Should be used when requesting a big set of data (GET /customer-group or POST /search/customer-group if filter or sort is set).
• bulk_upsert(objs: list[ModelClass] | list[dict[str, Any] creates/updates multiple objects. Does always return dict of plain response. (POST /_action/sync)
• bulk_delete(objs: list[ModelClass] | list[dict[str, Any] deletes multiple objects. Does always return dict or plain response. (POST /_action/sync)

Not all functions are available for the StoreClient-Endpoints. But some of them have some additional functions. StoreClient-Endpoints using the base.StoreSearchEndpoint can use most of the filter functions, but not create/update/delete.

Filter

The filter() functions allows you to filter the result of an query. The parameters are basically the field names. You can add an appendix to change the filter type. Without it looks for direct matches (equals). The following appendices are available:

• __in expects a list of values, matches if the value is provided in this list (equalsAny)
• __contains matches values that contain this value (contains)
• __gt greater than (range)
• __gte greater than equal (range)
• __lt lower than (range)
• __lte lower than equal (range)
• __range expects a touple of two items, matches everything inbetween. inclusive. (range)
• __startswith matches if the value starts with this (prefix)
• __endswith matches if the value ends with this (suffix)

For some fields (that are returned as dict, like custom_fields) it's also possible to filter over the values of it's keys. To do so you can append the key seperated by "__" For example if we have a custom field called "preferred_protein" we can filter on it like this:

customer = await Customer.using(client=client).filter(custom_field__preferred_protein="fish")

# or with filter-type-appendix
customer = await Customer.using(client=client).filter(custom_field__preferred_protein__in=["fish", "chicken"])

Query Result Caching

You can cache query results for a specified duration using the cache_for parameter. This is useful when you're repeatedly fetching the same data within a short time window:

from shopware_api_client.endpoints.admin import CustomEntity

# Cache results for 300 seconds (5 minutes)
async for custom_entity in client.custom_entity.iter(cache_for=300):
    # Process custom_entity
    pass

# Or with other endpoint methods
customers = await client.customer.all(cache_for=60)  # Cache for 60 seconds
customer = await client.customer.first(cache_for=300)  # Cache for 300 seconds

The cached results will be stored using your configured cache backend (either the default in-memory DictCache or a custom implementation). The TTL (time-to-live) is specified in seconds. Once the TTL expires, the next query will fetch fresh data from the API.

ApiModelBase

The base.ApiModelBase class is basicly a pydantic.BaseModel which should be used to create Endpoint-Models.

The base structure of an Endpoint-Model looks like this. Field names are converted to snake_case. Aliases are autogenerated:

from pydantic import Field
from typing import Any
from shopware_api_client.base import ApiModelBase, CustomFieldsMixin


class CustomerGroup(ApiModelBase, CustomFieldsMixin):
    _identifier = "customer_group"  # name of the Shopware-Endpoint (snaky)

    name: str  # Field with type
    display_gross: bool | None = None
    # other fields...

This Base-Models live in shopware_api_client.models

The id, version_id, created_at, updated_at and translated attributes are provided in the ApiModelBase and must not be added. This are default fields of Shopwares Entity class, even they are not always used.

If an entity supports the EntityCustomFieldsTrait you can add the CustomFieldsMixin to add the custom_fields field.

List of available Function

• save() executes Endpoint.update() if an id is set otherwise Endpoint.create()
• delete() executes Endpoint.delete()

AdminModel + Relations

To make relations to other models work, we have to define them in the Model. There are two classes to make this work: endpoints.relations.ForeignRelation and endpoints.relations.ManyRelation.

• ForeignRelation[class] is used when we get the id of the related object in the api response.

  • class: Class of the related model

• ManyRelation[class] is used for the reverse relation. We don't get ids in the api response, but it can be used through relation links.

  • class: Class of the related model

Example (Customer):

from pydantic import Field

from shopware_api_client.base import AdminModel
from shopware_api_client.endpoints.base_fields import IdField
from shopware_api_client.endpoints.relations import ForeignRelation, ManyRelation
from shopware_api_client.models.customer import Customer as CustomerBase

"""
// shopware_api_client.models.customer.Customer:
class Customer(ApiModelBase):
    # we have an id so we can create a ForeignRelation to it
    default_billing_address_id: IdField
"""

# final model containing relations for admin api. Must be AdminModel
class Customer(CustomerBase, AdminModel["CustomerEndpoint"]):
    default_billing_address: ForeignRelation["CustomerAddress"]

    # We don't have a field for all addresses of a customer, but there is a relation for it!
    addresses: ManyRelation["CustomerAddress"]

# model relation classes have to be imported at the end. pydantic needs the full import (not just TYPE_CHECKING)
# and this saves us from circular imports
from shopware_api_client.endpoints.admin import CustomerAddress  # noqa: E402

Development

Testing

You can use uv sync to install dependencies, and then run uv run pytest to execute the tests.

uv sync
uv run pytest

Model Creation

Shopware provides API-definitions for their whole API. You can download it from <shopurl>/api/_info/openapi3.json Then you can use tools like datamodel-code-generator

datamodel-codegen --input openapi3.json --output model_openapi3.py --snake-case-field --use-double-quotes --output-model-type=pydantic_v2.BaseModel --use-standard-collections --use-union-operator

The file may look confusing at first, but you can search for Endpoint-Name + JsonApi (Example: class CustomerJsonApi) to get all returned fields + relationships class as an overview over the available Relations. However, the Models will need some modifications. But it's a good start.

Not all fields returned by the API are writeable and the API will throw an error when you try to set it. So this fields must have an exclude=True in their definition. To find out which fields need to be excluded check the Shopware Endpoint documentation at https://shopware.stoplight.io/. Go to the Endpoint your Model belongs to and check the available POST fields.

The newly created Model and its Endpoint must than be imported to admin/__init__.py or store/__init__.py. The Model must be added to __all__ The Endpoint must be added to the Endpoints class. The __all__ statement is necessary so they don't get cleaned away as unused imports by code-formaters/cleaners.

We need to import all related models at the end of the file. If we don't add them, Pydantic fails to build the model. If we add them before our model definition, we run into circular imports.

Every model has a base model that lives in models/<model_file>.py. This model only contains direct fields (no relations), which are used by both endpoints (store & admin). This base models are extended in the endpoints to contain relations (endpoints/admin/core/<model_file>.py & endpoints/store/core/<model_file>.py).

Structure

> endpoints  -- All endpoints live here
  > admin  -- AdminAPI endpoints
    > core  -- AdminAPI > Core
      customer_address.py  -- Every Endpoint has its own file. Model and Endpoint are defined here
    > commercial  -- AdminAPI > Commercial
    > digital_sales_rooms  -- AdminAPI > Digital Sales Rooms
  > store  -- StoreAPI
    > core  -- StoreAPI > Core
    > commercial  -- StoreAPI > Commercial
    > digital_sales_rooms  -- StoreAPI > Digital Sales Rooms
> models  -- base models to be extended in admin/store endpoints (shopware => Entity)
> structs  -- data structures that do not have a model (shopware => Struct)
base.py  -- All the Base Classes (nearly)
fieldsets.py  -- FieldSetBase has its own file to prevent pydantic problems
client.py  -- Clients & Registry
config.py  -- Configs
exceptions.py  -- Exceptions
logging.py  -- Logging
tests.py  -- tests