Auth Bearer Token Caching
Reusing client_credential bearer tokens for the life of its expiration.
OAuth 2.0 Bearer Token Caching Best Practices
This guide covers best practices for caching and reusing OAuth 2.0 bearer tokens when integrating with Snapdocs APIs. Implementing token caching reduces latency, minimizes load on the authorization server, and improves the overall reliability of your integration.
Overview
Snapdocs APIs use OAuth 2.0 with the client_credentials grant type. When you request a token, the response includes an expires_in field indicating how long the token remains valid (typically 2 hours / 7200 seconds). By caching the token and reusing it for its full lifetime, your application avoids the overhead of fetching a new token on every API call.
Benefits of Token Caching
| Concern | Without Caching | With Caching |
|---|---|---|
| Latency per API call | +200–500ms (token fetch round-trip) | ~0ms (in-memory or cache lookup) |
| Auth server load | 1 token request per API call | 1 token request per ~2 hours |
| Rate limit risk | Higher — more requests to auth server | Negligible |
| Reliability | Additional point of failure per call | Token fetch failure is isolated |
How Token Caching Works
The following diagram illustrates the recommended flow. Your application checks for a cached token before each API call and only requests a new one when the cache is empty or the token has expired.
sequenceDiagram
participant Client as Your Application
participant Cache as Token Cache (Redis/Memory)
participant Auth as Auth0 (Token Endpoint)
participant API as Snapdocs API
Client->>Cache: Check for cached token
Cache-->>Client: Cache MISS
Client->>Auth: POST /oauth/token (client_credentials)
Auth-->>Client: { access_token, expires_in: 7200 }
Client->>Cache: Store token (TTL = expires_in - 60s buffer)
Client->>API: API request with Bearer token
API-->>Client: 200 OK
Note over Client,Cache: Subsequent API call (within token lifetime)
Client->>Cache: Check for cached token
Cache-->>Client: Cache HIT ✅
Client->>API: API request with Bearer token
API-->>Client: 200 OK
Note over Client,Auth: Token fetched only when cache is empty or expired
Token Lifecycle Decision Flow
Use this decision diagram to determine when your application should fetch a new token vs. reuse the cached one.
flowchart TD
A[API Call Needed] --> B{Cached token exists?}
B -- No --> C[Fetch new token from Auth0]
C --> D[Cache token with TTL = expires_in - 60s]
D --> E[Make API call with Bearer token]
B -- Yes --> F{Token expired or near expiry?}
F -- No --> E
F -- Yes --> C
E --> G{Response 401 Unauthorized?}
G -- No --> H[Process response]
G -- Yes --> I[Invalidate cached token]
I --> C
Implementation Guide
Step 1: Store the Token with Expiration Metadata
When you receive a token response from Auth0, store both the access_token and its calculated expiration time.
Buffer TimeWe recommend subtracting a buffer (e.g., 60 seconds) from the
expires_invalue, which is in seconds. This ensures your application refreshes the token before it actually expires, avoiding failed API calls due to clock skew or in-flight request timing.
A typical Auth0 token response:
{
"access_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
"token_type": "Bearer",
"expires_in": 7200
}Step 2: Check Cache Before Every API Call
Before making any API call, check your cache for a valid token. Only fetch a new one if the cache is empty or the token has expired.
Step 3: Handle 401 Responses Gracefully
If the API returns a 401 Unauthorized, the token may have been revoked or is otherwise invalid. Invalidate your cached token, fetch a new one, and retry the request once.
Code Examples
Python
using System;
using System.Net.Http;
using System.Text;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
public class SnapdocsTokenManager
{
private const int BufferSeconds = 60;
private readonly string _clientId;
private readonly string _clientSecret;
private readonly string _audience;
private readonly string _tokenUrl;
private string _accessToken;
private DateTime _expiresAt = DateTime.MinValue;
private readonly SemaphoreSlim _semaphore = new(1, 1);
private static readonly HttpClient _httpClient = new();
public SnapdocsTokenManager(string clientId, string clientSecret,
string audience, string tokenUrl)
{
_clientId = clientId;
_clientSecret = clientSecret;
_audience = audience;
_tokenUrl = tokenUrl;
}
public async Task<string> GetTokenAsync()
{
if (IsTokenValid()) return _accessToken;
await _semaphore.WaitAsync();
try
{
if (IsTokenValid()) return _accessToken;
await FetchNewTokenAsync();
return _accessToken;
}
finally
{
_semaphore.Release();
}
}
public void Invalidate()
{
_accessToken = null;
_expiresAt = DateTime.MinValue;
}
private bool IsTokenValid() =>
_accessToken != null && DateTime.UtcNow < _expiresAt;
private async Task FetchNewTokenAsync()
{
var payload = new
{
client_id = _clientId,
client_secret = _clientSecret,
audience = _audience,
grant_type = "client_credentials"
};
var content = new StringContent(
JsonSerializer.Serialize(payload),
Encoding.UTF8, "application/json");
var response = await _httpClient.PostAsync(_tokenUrl, content);
response.EnsureSuccessStatusCode();
var json = JsonDocument.Parse(await response.Content.ReadAsStringAsync());
_accessToken = json.RootElement.GetProperty("access_token").GetString();
var expiresIn = json.RootElement.TryGetProperty("expires_in", out var exp)
? exp.GetInt32() : 7200;
_expiresAt = DateTime.UtcNow.AddSeconds(expiresIn - BufferSeconds);
}
}import java.net.URI;
import java.net.http.*;
import java.time.Instant;
import com.google.gson.JsonParser;
public class SnapdocsTokenManager {
private static final int BUFFER_SECONDS = 60;
private final String clientId;
private final String clientSecret;
private final String audience;
private final String tokenUrl;
private String accessToken;
private Instant expiresAt = Instant.EPOCH;
private final Object lock = new Object();
public SnapdocsTokenManager(String clientId, String clientSecret,
String audience, String tokenUrl) {
this.clientId = clientId;
this.clientSecret = clientSecret;
this.audience = audience;
this.tokenUrl = tokenUrl;
}
public String getToken() throws Exception {
if (isTokenValid()) return accessToken;
synchronized (lock) {
if (isTokenValid()) return accessToken;
fetchNewToken();
return accessToken;
}
}
public void invalidate() {
synchronized (lock) {
accessToken = null;
expiresAt = Instant.EPOCH;
}
}
private boolean isTokenValid() {
return accessToken != null && Instant.now().isBefore(expiresAt);
}
private void fetchNewToken() throws Exception {
var body = String.format(
"{\"client_id\":\"%s\",\"client_secret\":\"%s\"," +
"\"audience\":\"%s\",\"grant_type\":\"client_credentials\"}",
clientId, clientSecret, audience
);
var request = HttpRequest.newBuilder()
.uri(URI.create(tokenUrl))
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(body))
.build();
var response = HttpClient.newHttpClient()
.send(request, HttpResponse.BodyHandlers.ofString());
var json = JsonParser.parseString(response.body()).getAsJsonObject();
accessToken = json.get("access_token").getAsString();
int expiresIn = json.has("expires_in") ? json.get("expires_in").getAsInt() : 7200;
expiresAt = Instant.now().plusSeconds(expiresIn - BUFFER_SECONDS);
}
}import time
import threading
import requests
class SnapdocsTokenManager:
"""
Thread-safe OAuth2 token manager with automatic caching and refresh.
Fetches a new token only when the cached token is expired or near expiry.
"""
BUFFER_SECONDS = 60 # Refresh 60s before actual expiry
def __init__(self, client_id, client_secret, audience, token_url):
self._client_id = client_id
self._client_secret = client_secret
self._audience = audience
self._token_url = token_url
self._access_token = None
self._expires_at = 0
self._lock = threading.Lock()
def get_token(self):
"""Return a valid access token, fetching a new one if needed."""
if self._is_token_valid():
return self._access_token
with self._lock:
# Double-check after acquiring lock
if self._is_token_valid():
return self._access_token
self._fetch_new_token()
return self._access_token
def invalidate(self):
"""Force token refresh on next call (e.g., after a 401)."""
with self._lock:
self._access_token = None
self._expires_at = 0
def _is_token_valid(self):
return (
self._access_token is not None
and time.time() < self._expires_at
)
def _fetch_new_token(self):
response = requests.post(self._token_url, json={
'client_id': self._client_id,
'client_secret': self._client_secret,
'audience': self._audience,
'grant_type': 'client_credentials',
})
response.raise_for_status()
data = response.json()
self._access_token = data['access_token']
expires_in = data.get('expires_in', 7200)
self._expires_at = time.time() + expires_in - self.BUFFER_SECONDS
# --- Usage ---
token_manager = SnapdocsTokenManager(
client_id='YOUR_CLIENT_ID',
client_secret='YOUR_CLIENT_SECRET',
audience='https://api.example.com',
token_url='https://auth.example.com/oauth/token',
)
# Every API call reuses the cached token
headers = {'Authorization': f'Bearer {token_manager.get_token()}'}
response = requests.get('https://api.example.com/v1/resource/123', headers=headers)
# If you receive a 401, invalidate and retry
if response.status_code == 401:
token_manager.invalidate()
headers = {'Authorization': f'Bearer {token_manager.get_token()}'}
response = requests.get('https://api.example.com/v1/resource/123', headers=headers)require 'faraday'
require 'json'
class SnapdocsTokenManager
BUFFER_SECONDS = 60 # Refresh 60s before actual expiry
def initialize(client_id:, client_secret:, audience:, token_url:)
@client_id = client_id
@client_secret = client_secret
@audience = audience
@token_url = token_url
@access_token = nil
@expires_at = Time.at(0)
@mutex = Mutex.new
end
def get_token
return @access_token if token_valid?
@mutex.synchronize do
return @access_token if token_valid?
fetch_new_token
@access_token
end
end
def invalidate!
@mutex.synchronize do
@access_token = nil
@expires_at = Time.at(0)
end
end
private
def token_valid?
!@access_token.nil? && Time.now < @expires_at
end
def fetch_new_token
conn = Faraday.new(url: @token_url)
response = conn.post do |req|
req.headers['Content-Type'] = 'application/json'
req.body = {
client_id: @client_id,
client_secret: @client_secret,
audience: @audience,
grant_type: 'client_credentials'
}.to_json
end
raise "Token fetch failed: #{response.status}" unless response.success?
data = JSON.parse(response.body)
@access_token = data['access_token']
expires_in = (data['expires_in'] || 7200).to_i
@expires_at = Time.now + expires_in - BUFFER_SECONDS
end
end
# --- Usage ---
token_manager = SnapdocsTokenManager.new(
client_id: 'YOUR_CLIENT_ID',
client_secret: 'YOUR_CLIENT_SECRET',
audience: 'https://api.example.com',
token_url: 'https://auth.example.com/oauth/token'
)
# Every API call reuses the cached token
response = Faraday.get('https://api.example.com/v1/resource/123') do |req|
req.headers['Authorization'] = "Bearer #{token_manager.get_token}"
req.headers['Content-Type'] = 'application/json'
end
# If you receive a 401, invalidate and retry
if response.status == 401
token_manager.invalidate!
response = Faraday.get('https://api.example.com/v1/resource/123') do |req|
req.headers['Authorization'] = "Bearer #{token_manager.get_token}"
req.headers['Content-Type'] = 'application/json'
end
endChoosing a Cache Backend
Any of the following approaches will work. Choose the one that best fits your existing infrastructure:
| Priority | Approach | Best For | Trade-offs |
|---|---|---|---|
| 1️⃣ | Redis / Memcached | Multi-instance or distributed services | Shared across all instances; requires cache infrastructure |
| 2️⃣ | In-memory (singleton) | Single-instance apps, serverless with warm starts | Simple to implement; lost on restart; not shared across instances |
| 3️⃣ | Database | When neither Redis nor in-memory caching is available | Works with any existing database; slightly higher latency per lookup |
Use What You HaveIf your application already has Redis or Memcached, that's the ideal choice — the token is shared across all instances and survives restarts. If not, an in-memory singleton is the simplest option and works well for single-instance apps. If neither is readily available, storing the token in a database table with an expiration timestamp is a perfectly valid approach — the small overhead of a database read is far less than fetching a new token from Auth0 on every call.
Multi-Service / Distributed Architecture
If your integration spans multiple services that each call Snapdocs APIs, consider a centralized token service:
flowchart LR
subgraph Your Infrastructure
SA[Service A] --> TS[Token Service / Cache]
SB[Service B] --> TS
SC[Service C] --> TS
TS --- Redis[(Redis / Shared Cache)]
end
TS -->|Fetch on cache miss| Auth[Auth0 Token Endpoint]
SA -->|Bearer Token| API[Snapdocs API]
SB -->|Bearer Token| API
SC -->|Bearer Token| API
Estimated Implementation Timeline
| Task | Estimate |
|---|---|
Implement TokenManager class with in-memory cache | 1–2 hours |
Replace direct token fetch calls with TokenManager | 1–2 hours |
| Add 401 retry logic with token invalidation | 1 hour |
| Unit tests for caching, expiry, and thread safety | 2–3 hours |
| Integration / end-to-end validation | 1–2 hours |
| Total | ~1 day |
Minimal Architectural ChangeToken caching is a client-side only change. No modifications are needed on the Snapdocs side. The change is isolated to how your application manages its bearer token — wrap your existing token-fetch logic in a caching layer and reuse the token for its full lifetime.
Summary
- Fetch a token once from Auth0 using the
client_credentialsgrant - Cache the token with a TTL of
expires_in - 60 seconds(buffer for clock skew) - Reuse the cached token for all API calls within the TTL window
- On 401 response, invalidate the cache and fetch a new token
- Thread safety — use a mutex/lock to prevent concurrent token fetches
Updated 6 days ago
