Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions docs/admin-api.md
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,46 @@ await OnlineToken.load(store_name, associated_user_id)
await PrivateToken.load(store_name)
```

### Client Credentials Grant Token

Starting January 2026, Shopify offline access tokens will expire after 24 hours. To refresh the token, use the `obtain_client_credentials_token` method on an `OfflineToken` instance.

[Read more about Client Credentials Grant](https://shopify.dev/docs/apps/build/authentication-authorization/access-tokens/client-credentials-grant)

```python
# Load an existing offline token
offline_token = await OfflineToken.load(store_name='my-store')

# OR ceate a new offline token
offline_token = OfflineToken(store_name='my-store')

# Obtain a new access token using client credentials
await offline_token.obtain_client_credentials_token(
client_id='your_api_key',
client_secret='your_api_secret_key'
)

# The token is now updated with the new access_token and expiry
print(offline_token.access_token)
# Don't forget to save it
await offline_token.save()
```

After calling `obtain_client_credentials_token`, the token instance will have:
- A new `access_token`
- An updated `expires_unix_timestamp` (typically 24 hours from now)

You can check if a token has expired using:

```python
if offline_token.is_expired():
await offline_token.obtain_client_credentials_token(
client_id='your_api_key',
client_secret='your_api_secret_key'
)
await offline_token.save()
```

## Querying Shopify

### REST
Expand Down
41 changes: 39 additions & 2 deletions spylib/admin_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
MAX_COST_EXCEEDED_ERROR_CODE,
OPERATION_NAME_REQUIRED_ERROR_MESSAGE,
THROTTLED_ERROR_CODE,
TOKEN_EXPIRATION_BUFFER_SECONDS,
WRONG_OPERATION_NAME_ERROR_MESSAGE,
)
from spylib.exceptions import (
Expand All @@ -32,7 +33,8 @@
ShopifyThrottledError,
not_our_fault,
)
from spylib.utils.misc import TimedResult, elapsed_time, parse_scope
from spylib.oauth.models import ClientCredentialsTokenModel
from spylib.utils.misc import TimedResult, elapsed_time, now_epoch, parse_scope
from spylib.utils.rest import Request


Expand Down Expand Up @@ -62,6 +64,8 @@ class Token(ABC, BaseModel):

client: ClassVar[AsyncClient] = AsyncClient()

expires_unix_timestamp: Optional[int] = None

@property
def oauth_url(self) -> str:
return f'https://{self.store_name}.myshopify.com/admin/oauth/access_token'
Expand Down Expand Up @@ -114,6 +118,16 @@ async def __handle_error(self, debug: str, endpoint: str, response: Response):

raise ShopifyError(msg)

def set_expires_unix_timestamp(self, expires_in: int):
# unix timestamp
self.expires_unix_timestamp = now_epoch() + expires_in

def is_expired(self) -> bool:
"""Check if the token is expired."""
if not self.expires_unix_timestamp:
raise ValueError('Token does not have an expiry time')
return now_epoch() - self.expires_unix_timestamp > TOKEN_EXPIRATION_BUFFER_SECONDS

@retry(
reraise=True,
wait=wait_random(min=1, max=2),
Expand Down Expand Up @@ -279,7 +293,11 @@ async def test_connection(self) -> bool:


class OfflineTokenABC(Token, ABC):
"""Offline tokens are used for long term access, and do not have a set expiry."""
"""Offline tokens are used for long term access, do not have a set expiry prior to 2026-01-01.

All newly supported access tokens will have an expiry of 24 hours after 2026-01-01.
[Read more about it](https://shopify.dev/docs/apps/build/authentication-authorization/access-tokens/client-credentials-grant).
"""

@abstractmethod
async def save(self):
Expand All @@ -290,6 +308,25 @@ async def save(self):
async def load(cls, store_name: str):
pass

async def obtain_client_credentials_token(self, client_id: str, client_secret: str) -> None:
response = await self.client.post(
url=self.oauth_url,
data={
'grant_type': 'client_credentials',
'client_id': client_id,
'client_secret': client_secret,
},
)
if response.status_code != status.HTTP_200_OK:
raise ShopifyError(
f'Failed to obtain client credentials token: {response.status_code}'
)

client_credentials_token = ClientCredentialsTokenModel.model_validate(response.json())
self.set_expires_unix_timestamp(client_credentials_token.expires_in)
self.access_token = client_credentials_token.access_token
self.scope = client_credentials_token.scope.split(',')


class OnlineTokenABC(Token, ABC):
"""Online tokens are used to implement applications authenticated with a specific user's credentials.
Expand Down
2 changes: 2 additions & 0 deletions spylib/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,3 +11,5 @@

UTF8ENCODING = 'utf-8'
API_CALL_NUMBER_RETRY_ATTEMPTS = 5

TOKEN_EXPIRATION_BUFFER_SECONDS = 300
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@lishanl What does this do?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@lishanl What does this do?

Consider the token expired 5 min before the actual expiration (when used in the check) - a pretty generous time buffer for the calculation to detect token expiration for any renewal

12 changes: 12 additions & 0 deletions spylib/oauth/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,3 +84,15 @@ class OnlineTokenModel(BaseModel):

associated_user: AssociatedUser
"""The Shopify user associated with this token."""


class ClientCredentialsTokenModel(BaseModel):
"""[Read more about Client credentials grant](https://shopify.dev/docs/apps/build/authentication-authorization/access-tokens/client-credentials-grant)."""

access_token: str
"""An API access token that can be used to access the shop's data until it expires or revoked."""

expires_in: int
"""The number of seconds until this session (and `access_token`) expire."""

scope: str
3 changes: 2 additions & 1 deletion spylib/oauth/validations.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

from spylib.hmac import validate as validate_hmac

from ..constants import TOKEN_EXPIRATION_BUFFER_SECONDS
from ..utils import domain_to_storename, now_epoch
from .tokens import OAuthJWT

Expand All @@ -13,7 +14,7 @@ def validate_callback(shop: str, timestamp: int, query_string: Any, api_secret_k
domain_to_storename(shop)

# 2) Check the timestamp. Must not be more than 5min old
if now_epoch() - timestamp > 300:
if now_epoch() - timestamp > TOKEN_EXPIRATION_BUFFER_SECONDS:
raise ValueError('Timestamp is too old')

# 3) Check the hmac
Expand Down
26 changes: 26 additions & 0 deletions tests/oauth/test_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@
from spylib.exceptions import FastAPIImportError
from spylib.utils import JWTBaseModel, domain_to_storename, now_epoch, store_domain

from ..token_classes import OfflineToken

SHOPIFY_API_KEY = 'API_KEY'
SHOPIFY_SECRET_KEY = 'SECRET_KEY'

Expand Down Expand Up @@ -48,6 +50,12 @@
},
)

CLIENTCREDENTIALSTOKEN_DATA = dict(
access_token='CLIENTCREDENTIALSTOKEN',
expires_in=86399,
scope='write_products,read_customers,write_orders',
)


@dataclass
class MockHTTPResponse:
Expand Down Expand Up @@ -212,6 +220,24 @@ def check_oauth_redirect_query(query: str, scope: List[str], query_extra: dict =
return state


@pytest.mark.asyncio
async def test_oauth_client_credentials_token(mocker):
# test offline token obtain_client_credentials_token using the client credentials token data
mocker.patch(
'httpx.AsyncClient.post',
return_value=MockHTTPResponse(status_code=200, jsondata=CLIENTCREDENTIALSTOKEN_DATA),
)
offline_token = await OfflineToken.load(store_name=TEST_STORE)
await offline_token.obtain_client_credentials_token(
client_id=SHOPIFY_API_KEY, client_secret=SHOPIFY_SECRET_KEY
)
assert offline_token.access_token == CLIENTCREDENTIALSTOKEN_DATA['access_token']
assert offline_token.expires_unix_timestamp is not None
# approximate comparison for 24 hours in seconds after it's obtained without timefreeze
assert offline_token.expires_unix_timestamp > now_epoch() + 86300
assert offline_token.is_expired() is False


def test_domain_to_storename():
assert domain_to_storename(domain='test.myshopify.com') == 'test'
assert domain_to_storename(domain='test2.myshopify.com') == 'test2'
Expand Down