| Name | sellerlegend-api JSON |
| Version |
1.0.3
JSON |
| download |
| home_page | https://github.com/sellerlegend/sellerlegend_api_py |
| Summary | Official Python SDK for the SellerLegend API |
| upload_time | 2025-09-19 05:31:54 |
| maintainer | None |
| docs_url | None |
| author | SellerLegend |
| requires_python | >=3.8 |
| license | MIT License
Copyright (c) 2024 SellerLegend
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE. |
| keywords |
sellerlegend
api
amazon
seller
ecommerce
analytics
amazon-seller
marketplace
fba
seller-central
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
requests
requests-oauthlib
python-dateutil
typing-extensions
python-dotenv
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# SellerLegend Python SDK
The official Python SDK for the SellerLegend API, providing a comprehensive interface to integrate SellerLegend's Amazon seller analytics and management capabilities into your Python applications.
## Features
- **OAuth2 Authentication**: Full support for Laravel Passport OAuth2 authentication
- **Comprehensive API Coverage**: Access to all major SellerLegend API endpoints
- **Type Hints**: Full type annotations for better IDE support and code reliability
- **Error Handling**: Comprehensive exception hierarchy for different error types
- **Automatic Token Refresh**: Handles token expiration and refresh automatically
- **Request Retry Logic**: Built-in retry mechanism for resilient API communication
- **Async Report Downloads**: Support for asynchronous report generation and download
## Installation
```bash
pip install sellerlegend-api
```
Or install from source:
```bash
git clone git@github.com:sellerlegend/sellerlegend_api_py.git
cd sellerlegend_api_py
pip install -e .
```
## Quick Start
### Step 1: Obtain API Credentials
Before using the API, you need to register your application in SellerLegend:
1. Log in to your SellerLegend account
2. Navigate to **Admin → Developers**
3. Click on **"Create New Client"**
4. Fill in your application details:
- **Name:** Your application name
- **Redirect URL:** Your OAuth callback URL (e.g., `https://yourapp.com/callback`)
5. Click **"Create"** to generate your credentials
**Important:** Save your **Client ID** and **Client Secret** immediately. The Client Secret will only be shown once for security reasons.
### Step 2: OAuth2 Authorization Code Flow (The ONLY Method for Full API Access)
```python
from sellerlegend_api import SellerLegendClient
import time
# Initialize the client with credentials from Step 1
client = SellerLegendClient(
client_id="your_oauth_client_id", # From Admin → Developers
client_secret="your_oauth_client_secret", # From Admin → Developers
base_url="https://app.sellerlegend.com"
)
# Step 1: Get authorization URL
auth_url, state = client.get_authorization_url()
print(f"Please visit: {auth_url}")
print(f"State (save for verification): {state}")
# Step 2: User authorizes and you receive a code at your callback URL
code = input("Enter the authorization code: ")
# Step 3: Exchange code for tokens
token_info = client.authenticate_with_code(code, state)
# Step 4: CRITICAL - Store BOTH tokens in your database
store_in_database(
access_token=token_info['access_token'],
refresh_token=token_info['refresh_token'], # MUST store this!
expires_at=time.time() + token_info['expires_in']
)
print(f"Authentication successful! Token expires in {token_info['expires_in']} seconds")
```
## Authentication
⚠️ **CRITICAL INFORMATION**:
- **Only the OAuth 2.0 Authorization Code flow provides full API access**
- Personal Access Tokens are **NOT** supported
- Direct access tokens are **NOT** supported
- Password Grant is **NOT** supported
- Client Credentials Grant only provides access to service status endpoint
### 1. OAuth 2.0 Authorization Code Flow - Complete Implementation
This is the **ONLY** way to get full API access. You MUST:
1. Implement the complete OAuth flow
2. Store tokens securely in a database
3. Handle token refresh properly
4. **Update refresh tokens after each refresh** (old ones become invalid)
#### Step-by-Step Implementation:
```python
import time
import sqlite3
from sellerlegend_api import SellerLegendClient
class SellerLegendTokenManager:
"""Complete token management with database storage."""
def __init__(self, client_id, client_secret, db_path="tokens.db"):
self.client_id = client_id
self.client_secret = client_secret
self.base_url = "https://app.sellerlegend.com"
self.db_path = db_path
self.init_database()
def init_database(self):
"""Create tokens table if it doesn't exist."""
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS oauth_tokens (
id INTEGER PRIMARY KEY AUTOINCREMENT,
access_token TEXT NOT NULL UNIQUE,
refresh_token TEXT NOT NULL,
expires_at INTEGER NOT NULL,
created_at INTEGER NOT NULL,
updated_at INTEGER
)
""")
conn.commit()
conn.close()
def complete_oauth_flow(self, redirect_uri="http://localhost:5001/callback"):
"""Complete OAuth authorization flow."""
# Initialize client
client = SellerLegendClient(
client_id=self.client_id,
client_secret=self.client_secret,
base_url=self.base_url
)
# Step 1: Get authorization URL
auth_url, state = client.get_authorization_url(redirect_uri)
print("=== OAuth Authorization Required ===")
print(f"1. Visit this URL: {auth_url}")
print(f"2. Log in and authorize the application")
print(f"3. You'll be redirected to: {redirect_uri}?code=AUTH_CODE&state={state}")
print("")
# Step 2: Get authorization code from user
code = input("Enter the authorization code from the redirect URL: ")
# Step 3: Exchange code for tokens
try:
token_info = client.authenticate_with_code(code, state)
# Step 4: Store tokens in database
self.store_tokens(token_info)
print("✅ Authentication successful!")
print(f"Access token expires in: {token_info['expires_in']} seconds")
print(f"Refresh token expires in: 30 days")
return token_info
except Exception as e:
print(f"❌ Authentication failed: {e}")
raise
def store_tokens(self, token_info):
"""
Store tokens in database.
CRITICAL: Must store BOTH access and refresh tokens!
"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
expires_at = int(time.time() + token_info['expires_in'])
created_at = int(time.time())
# Delete old tokens (optional, for single-user apps)
cursor.execute("DELETE FROM oauth_tokens")
# Insert new tokens
cursor.execute("""
INSERT INTO oauth_tokens
(access_token, refresh_token, expires_at, created_at)
VALUES (?, ?, ?, ?)
""", (
token_info['access_token'],
token_info['refresh_token'], # CRITICAL: Must store refresh token!
expires_at,
created_at
))
conn.commit()
conn.close()
print("✅ Tokens stored securely in database")
def get_valid_client(self):
"""
Get an authenticated client, refreshing token if needed.
"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Get most recent tokens
cursor.execute("""
SELECT * FROM oauth_tokens
ORDER BY created_at DESC
LIMIT 1
""")
tokens = cursor.fetchone()
conn.close()
if not tokens:
print("❌ No tokens found. Please run complete_oauth_flow() first.")
raise Exception("Not authenticated")
current_time = int(time.time())
# Check if token needs refresh (5 minute buffer)
if tokens['expires_at'] <= current_time + 300:
print("⚠️ Token expired or expiring soon, refreshing...")
return self.refresh_and_get_client(tokens['refresh_token'])
# Return client with valid token
return SellerLegendClient(
base_url=self.base_url,
access_token=tokens['access_token']
)
def refresh_and_get_client(self, refresh_token):
"""
Refresh tokens and return authenticated client.
⚠️ CRITICAL: When refreshing, you get a NEW refresh token!
The old refresh token becomes INVALID immediately!
"""
client = SellerLegendClient(
client_id=self.client_id,
client_secret=self.client_secret,
base_url=self.base_url
)
try:
# Get new tokens
new_tokens = client.refresh_access_token(refresh_token)
# CRITICAL: Store the NEW tokens (including refresh token!)
self.store_tokens(new_tokens)
print("✅ Tokens refreshed successfully")
print("⚠️ WARNING: Old refresh token is now INVALID!")
# Return client with new access token
return SellerLegendClient(
base_url=self.base_url,
access_token=new_tokens['access_token']
)
except Exception as e:
print(f"❌ Token refresh failed: {e}")
print("User must re-authenticate with OAuth flow")
raise
# Usage Example
if __name__ == "__main__":
# Initialize token manager
manager = SellerLegendTokenManager(
client_id="YOUR_CLIENT_ID",
client_secret="YOUR_CLIENT_SECRET"
)
# First time: Complete OAuth flow
# Uncomment this line for initial setup:
# manager.complete_oauth_flow()
# Subsequent uses: Get authenticated client
try:
client = manager.get_valid_client()
# Now you can make API calls
user = client.user.get_me()
print(f"Authenticated as: {user['user']['name']}")
# Get accounts
accounts = client.user.get_accounts()
for account in accounts['accounts']:
print(f"Account: {account['account_title']}")
# Get sales data
orders = client.sales.get_orders(
per_page=500,
start_date="2024-01-01",
end_date="2024-01-31"
)
print(f"Found {len(orders.get('data', []))} orders")
except Exception as e:
print(f"Error: {e}")
print("Please run complete_oauth_flow() to authenticate")
```
### 2. Token Refresh - Critical Information
⚠️ **CRITICAL POINTS ABOUT TOKEN REFRESH**:
1. **NEW Refresh Token**: When you refresh an access token, you receive a NEW refresh token
2. **Old Token Invalid**: The old refresh token becomes invalid IMMEDIATELY
3. **Must Update Storage**: You MUST update your stored refresh token every time
4. **Atomic Updates**: Use database transactions to prevent token loss
5. **Handle Failures**: If refresh fails, user must re-authenticate
```python
def refresh_tokens_safely(old_refresh_token, client_id, client_secret):
"""
Safe token refresh with proper error handling.
"""
client = SellerLegendClient(
client_id=client_id,
client_secret=client_secret,
base_url="https://app.sellerlegend.com"
)
try:
# Refresh tokens
new_tokens = client.refresh_access_token(old_refresh_token)
# CRITICAL: You now have NEW tokens
# The old refresh token is INVALID!
# Store BOTH new tokens immediately
with database.transaction(): # Use transaction for safety
database.execute("""
UPDATE oauth_tokens SET
access_token = ?,
refresh_token = ?, # NEW refresh token!
expires_at = ?,
updated_at = ?
WHERE refresh_token = ?
""", [
new_tokens['access_token'],
new_tokens['refresh_token'], # Must store the NEW one!
time.time() + new_tokens['expires_in'],
time.time(),
old_refresh_token # Match on old token
])
print("✅ Tokens refreshed and stored")
print("⚠️ Old refresh token is now invalid!")
return new_tokens
except Exception as e:
# Refresh failed - tokens may be invalid
print(f"❌ Token refresh failed: {e}")
# Mark tokens as invalid in database
database.execute(
"UPDATE oauth_tokens SET expires_at = 0 WHERE refresh_token = ?",
[old_refresh_token]
)
# User must re-authenticate
raise Exception("Token refresh failed. User must re-authenticate.")
```
### 3. Common Pitfalls and Solutions
#### Pitfall 1: Not Storing Refresh Token
```python
# ❌ WRONG - Only storing access token
database.save(access_token=token_info['access_token'])
# ✅ CORRECT - Store both tokens
database.save(
access_token=token_info['access_token'],
refresh_token=token_info['refresh_token']
)
```
#### Pitfall 2: Not Updating Refresh Token After Refresh
```python
# ❌ WRONG - Not updating refresh token
new_tokens = client.refresh_access_token(old_refresh_token)
database.update(access_token=new_tokens['access_token']) # Missing refresh token!
# ✅ CORRECT - Update both tokens
new_tokens = client.refresh_access_token(old_refresh_token)
database.update(
access_token=new_tokens['access_token'],
refresh_token=new_tokens['refresh_token'] # Must update this too!
)
```
#### Pitfall 3: Using Expired Refresh Token
```python
# ❌ WRONG - Using old refresh token after refresh
tokens1 = client.refresh_access_token(refresh_token)
# ... later ...
tokens2 = client.refresh_access_token(refresh_token) # Will fail!
# ✅ CORRECT - Use the new refresh token
tokens1 = client.refresh_access_token(old_refresh_token)
# ... later ...
tokens2 = client.refresh_access_token(tokens1['refresh_token']) # Use new token
```
## API Usage Examples
Once you have a valid authenticated client, you can access all API endpoints:
### Sales Data
```python
# Get authenticated client
client = token_manager.get_valid_client()
# Get recent orders
orders = client.sales.get_orders(
per_page=500,
start_date="2023-12-01",
end_date="2023-12-31"
)
# Get sales statistics
stats = client.sales.get_statistics_dashboard(
view_by="product",
group_by="sku",
per_page=1000,
currency="USD"
)
# Get daily sales per product
daily_sales = client.sales.get_per_day_per_product(
per_page=500,
start_date="2023-12-01",
end_date="2023-12-31"
)
# Get transaction data
transactions = client.sales.get_transactions(
per_page=1000,
start_date="2023-12-01",
end_date="2023-12-31"
)
```
### Reports
```python
# Create and download a report (async)
report_data = client.reports.create_and_download_report(
product_sku="YOUR-SKU-123",
timeout=300 # Wait up to 5 minutes
)
# Save the report to file
with open("report.csv.gz", "wb") as f:
f.write(report_data)
# Or handle the process manually
# 1. Create report request
create_response = client.reports.create_report_request(
product_sku="YOUR-SKU-123"
)
report_id = create_response["report_id"]
# 2. Check status periodically
status = client.reports.get_report_status(report_id)
print(f"Report status: {status['status']}")
# 3. Download when ready
if status['status'] == 'done':
report_data = client.reports.download_report(report_id)
```
### Inventory Management
```python
# Get inventory list
inventory = client.inventory.get_list(
per_page=500,
velocity_start_date="2023-11-01",
velocity_end_date="2023-12-31"
)
# Filter by specific SKU
inventory_filtered = client.inventory.get_list(
per_page=100,
filter_by="sku",
filter_value="YOUR-SKU-123"
)
```
### Cost Management (COGS)
```python
# Get cost periods for a product
cost_periods = client.costs.get_cost_periods(sku="YOUR-SKU-123")
# Update cost periods
cost_data = [
{
"dates": {
"from_date": "2023-01-01",
"to_date": "2023-12-31"
},
"cost_elements": [
{
"cost_element": "Product Cost",
"provider": "Supplier Name",
"notes": "Updated cost",
"total_amount": 15.50,
"currency": "USD",
"conversion_rate": "1.00",
"units": 1,
"amount": 15.50
}
]
}
]
result = client.costs.update_cost_periods(
sku="YOUR-SKU-123",
data=cost_data
)
```
### Supply Chain
```python
# Get restock suggestions
suggestions = client.supply_chain.get_restock_suggestions(
per_page=500,
currency="USD"
)
for suggestion in suggestions['data']:
print(f"SKU: {suggestion['sku']}, Suggested Reorder: {suggestion['suggested_quantity']}")
```
### Connections Status
```python
# Check Amazon connection status
connections = client.connections.get_list()
for conn in connections['connections']:
print(f"Account: {conn['account_title']}")
print(f"SP API Status: {conn['sp']['status']}")
print(f"PPC Status: {conn['ppc']['status']}")
```
## Account Filtering
Most endpoints support filtering by specific accounts or account groups:
```python
# Filter by account title
orders = client.sales.get_orders(
account_title="My Store",
per_page=500
)
# Filter by seller ID and marketplace
orders = client.sales.get_orders(
seller_id="A1SELLER123",
marketplace_id="ATVPDKIKX0DER",
per_page=500
)
# Filter by account group
stats = client.sales.get_statistics_dashboard(
view_by="product",
group_by="sku",
group_title="US Stores",
per_page=1000
)
```
## Error Handling
```python
from sellerlegend_api import (
SellerLegendClient,
AuthenticationError,
ValidationError,
NotFoundError,
RateLimitError,
ServerError
)
try:
# Get authenticated client
client = token_manager.get_valid_client()
# Make API calls
orders = client.sales.get_orders(per_page=500)
except AuthenticationError as e:
print(f"Authentication failed: {e.message}")
# Token may be expired, try refresh or re-authenticate
except ValidationError as e:
print(f"Validation error: {e.message}")
print(f"Response data: {e.response_data}")
except NotFoundError as e:
print(f"Resource not found: {e.message}")
except RateLimitError as e:
print(f"Rate limit exceeded: {e.message}")
# Implement backoff strategy
except ServerError as e:
print(f"Server error: {e.message}")
```
## Configuration
### Timeout and Retry Settings
```python
client = SellerLegendClient(
client_id="your_client_id",
client_secret="your_client_secret",
base_url="https://app.sellerlegend.com",
timeout=60, # 60 second timeout
max_retries=5, # Retry up to 5 times
backoff_factor=0.5 # Backoff factor for retries
)
```
## Available Resources
The SDK provides access to the following SellerLegend API resources:
- **`client.user`**: User information and account management
- **`client.sales`**: Orders, statistics, and transaction data
- **`client.reports`**: Report generation and download
- **`client.inventory`**: Inventory levels and management
- **`client.costs`**: Cost of goods sold (COGS) management
- **`client.connections`**: Amazon connection status
- **`client.supply_chain`**: Restock suggestions
- **`client.warehouse`**: Warehouse and inbound shipment data
- **`client.notifications`**: Notification management
## Rate Limiting
The SDK includes built-in retry logic for rate-limited requests. When a rate limit is encountered (HTTP 429), the client will automatically retry with exponential backoff.
## Important Security Notes
1. **Never commit credentials**: Keep your `client_id` and `client_secret` secure
2. **Use environment variables**: Store sensitive data in `.env` files
3. **Secure token storage**: Always store tokens in a database, never in files or cookies
4. **Use HTTPS**: Always use HTTPS for callbacks and API calls
5. **Validate state parameter**: Always validate the state parameter to prevent CSRF attacks
## Database Schema Example
```sql
-- Recommended schema for token storage
CREATE TABLE oauth_tokens (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER, -- Your application's user ID
access_token TEXT NOT NULL,
refresh_token TEXT NOT NULL,
expires_at INTEGER NOT NULL,
created_at INTEGER NOT NULL,
updated_at INTEGER,
UNIQUE(access_token)
);
-- Index for faster lookups
CREATE INDEX idx_user_id ON oauth_tokens(user_id);
CREATE INDEX idx_expires_at ON oauth_tokens(expires_at);
```
## Troubleshooting
### "Unauthenticated" Error
- Token may be expired - check `expires_at` timestamp
- Try refreshing the token using the refresh token
- If refresh fails, user must re-authenticate
### "Access Denied" Error
- User may not have API access enabled
- Check account permissions in SellerLegend
### Token Refresh Fails
- Refresh token may be expired (30 days)
- Refresh token may have been used already (they're single-use)
- User must complete OAuth flow again
### Rate Limiting
- Implement exponential backoff
- Cache frequently accessed data
- Use batch endpoints where available
## Support
For support, please contact [support@sellerlegend.com](mailto:support@sellerlegend.com) or visit our [documentation](https://docs.sellerlegend.com/api-docs/index.html).
## License
This SDK is licensed under the MIT License. See LICENSE file for details.
Raw data
{
"_id": null,
"home_page": "https://github.com/sellerlegend/sellerlegend_api_py",
"name": "sellerlegend-api",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "SellerLegend <support@sellerlegend.com>",
"keywords": "sellerlegend, api, amazon, seller, ecommerce, analytics, amazon-seller, marketplace, fba, seller-central",
"author": "SellerLegend",
"author_email": "SellerLegend <support@sellerlegend.com>",
"download_url": "https://files.pythonhosted.org/packages/94/1f/4de7d9d5e53e8110fa264e87e84b6f735839c5de1c678518eabee36053c0/sellerlegend_api-1.0.3.tar.gz",
"platform": null,
"description": "# SellerLegend Python SDK\n\nThe official Python SDK for the SellerLegend API, providing a comprehensive interface to integrate SellerLegend's Amazon seller analytics and management capabilities into your Python applications.\n\n## Features\n\n- **OAuth2 Authentication**: Full support for Laravel Passport OAuth2 authentication\n- **Comprehensive API Coverage**: Access to all major SellerLegend API endpoints\n- **Type Hints**: Full type annotations for better IDE support and code reliability\n- **Error Handling**: Comprehensive exception hierarchy for different error types\n- **Automatic Token Refresh**: Handles token expiration and refresh automatically\n- **Request Retry Logic**: Built-in retry mechanism for resilient API communication\n- **Async Report Downloads**: Support for asynchronous report generation and download\n\n## Installation\n\n```bash\npip install sellerlegend-api\n```\n\nOr install from source:\n\n```bash\ngit clone git@github.com:sellerlegend/sellerlegend_api_py.git\ncd sellerlegend_api_py\npip install -e .\n```\n\n## Quick Start\n\n### Step 1: Obtain API Credentials\n\nBefore using the API, you need to register your application in SellerLegend:\n\n1. Log in to your SellerLegend account\n2. Navigate to **Admin \u2192 Developers**\n3. Click on **\"Create New Client\"**\n4. Fill in your application details:\n - **Name:** Your application name\n - **Redirect URL:** Your OAuth callback URL (e.g., `https://yourapp.com/callback`)\n5. Click **\"Create\"** to generate your credentials\n\n**Important:** Save your **Client ID** and **Client Secret** immediately. The Client Secret will only be shown once for security reasons.\n\n### Step 2: OAuth2 Authorization Code Flow (The ONLY Method for Full API Access)\n\n```python\nfrom sellerlegend_api import SellerLegendClient\nimport time\n\n# Initialize the client with credentials from Step 1\nclient = SellerLegendClient(\n client_id=\"your_oauth_client_id\", # From Admin \u2192 Developers\n client_secret=\"your_oauth_client_secret\", # From Admin \u2192 Developers\n base_url=\"https://app.sellerlegend.com\"\n)\n\n# Step 1: Get authorization URL\nauth_url, state = client.get_authorization_url()\nprint(f\"Please visit: {auth_url}\")\nprint(f\"State (save for verification): {state}\")\n\n# Step 2: User authorizes and you receive a code at your callback URL\ncode = input(\"Enter the authorization code: \")\n\n# Step 3: Exchange code for tokens\ntoken_info = client.authenticate_with_code(code, state)\n\n# Step 4: CRITICAL - Store BOTH tokens in your database\nstore_in_database(\n access_token=token_info['access_token'],\n refresh_token=token_info['refresh_token'], # MUST store this!\n expires_at=time.time() + token_info['expires_in']\n)\n\nprint(f\"Authentication successful! Token expires in {token_info['expires_in']} seconds\")\n```\n\n## Authentication\n\n\u26a0\ufe0f **CRITICAL INFORMATION**:\n- **Only the OAuth 2.0 Authorization Code flow provides full API access**\n- Personal Access Tokens are **NOT** supported\n- Direct access tokens are **NOT** supported\n- Password Grant is **NOT** supported\n- Client Credentials Grant only provides access to service status endpoint\n\n### 1. OAuth 2.0 Authorization Code Flow - Complete Implementation\n\nThis is the **ONLY** way to get full API access. You MUST:\n1. Implement the complete OAuth flow\n2. Store tokens securely in a database\n3. Handle token refresh properly\n4. **Update refresh tokens after each refresh** (old ones become invalid)\n\n#### Step-by-Step Implementation:\n\n```python\nimport time\nimport sqlite3\nfrom sellerlegend_api import SellerLegendClient\n\nclass SellerLegendTokenManager:\n \"\"\"Complete token management with database storage.\"\"\"\n\n def __init__(self, client_id, client_secret, db_path=\"tokens.db\"):\n self.client_id = client_id\n self.client_secret = client_secret\n self.base_url = \"https://app.sellerlegend.com\"\n self.db_path = db_path\n self.init_database()\n\n def init_database(self):\n \"\"\"Create tokens table if it doesn't exist.\"\"\"\n conn = sqlite3.connect(self.db_path)\n conn.execute(\"\"\"\n CREATE TABLE IF NOT EXISTS oauth_tokens (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n access_token TEXT NOT NULL UNIQUE,\n refresh_token TEXT NOT NULL,\n expires_at INTEGER NOT NULL,\n created_at INTEGER NOT NULL,\n updated_at INTEGER\n )\n \"\"\")\n conn.commit()\n conn.close()\n\n def complete_oauth_flow(self, redirect_uri=\"http://localhost:5001/callback\"):\n \"\"\"Complete OAuth authorization flow.\"\"\"\n # Initialize client\n client = SellerLegendClient(\n client_id=self.client_id,\n client_secret=self.client_secret,\n base_url=self.base_url\n )\n\n # Step 1: Get authorization URL\n auth_url, state = client.get_authorization_url(redirect_uri)\n\n print(\"=== OAuth Authorization Required ===\")\n print(f\"1. Visit this URL: {auth_url}\")\n print(f\"2. Log in and authorize the application\")\n print(f\"3. You'll be redirected to: {redirect_uri}?code=AUTH_CODE&state={state}\")\n print(\"\")\n\n # Step 2: Get authorization code from user\n code = input(\"Enter the authorization code from the redirect URL: \")\n\n # Step 3: Exchange code for tokens\n try:\n token_info = client.authenticate_with_code(code, state)\n\n # Step 4: Store tokens in database\n self.store_tokens(token_info)\n\n print(\"\u2705 Authentication successful!\")\n print(f\"Access token expires in: {token_info['expires_in']} seconds\")\n print(f\"Refresh token expires in: 30 days\")\n\n return token_info\n\n except Exception as e:\n print(f\"\u274c Authentication failed: {e}\")\n raise\n\n def store_tokens(self, token_info):\n \"\"\"\n Store tokens in database.\n CRITICAL: Must store BOTH access and refresh tokens!\n \"\"\"\n conn = sqlite3.connect(self.db_path)\n cursor = conn.cursor()\n\n expires_at = int(time.time() + token_info['expires_in'])\n created_at = int(time.time())\n\n # Delete old tokens (optional, for single-user apps)\n cursor.execute(\"DELETE FROM oauth_tokens\")\n\n # Insert new tokens\n cursor.execute(\"\"\"\n INSERT INTO oauth_tokens\n (access_token, refresh_token, expires_at, created_at)\n VALUES (?, ?, ?, ?)\n \"\"\", (\n token_info['access_token'],\n token_info['refresh_token'], # CRITICAL: Must store refresh token!\n expires_at,\n created_at\n ))\n\n conn.commit()\n conn.close()\n\n print(\"\u2705 Tokens stored securely in database\")\n\n def get_valid_client(self):\n \"\"\"\n Get an authenticated client, refreshing token if needed.\n \"\"\"\n conn = sqlite3.connect(self.db_path)\n conn.row_factory = sqlite3.Row\n cursor = conn.cursor()\n\n # Get most recent tokens\n cursor.execute(\"\"\"\n SELECT * FROM oauth_tokens\n ORDER BY created_at DESC\n LIMIT 1\n \"\"\")\n\n tokens = cursor.fetchone()\n conn.close()\n\n if not tokens:\n print(\"\u274c No tokens found. Please run complete_oauth_flow() first.\")\n raise Exception(\"Not authenticated\")\n\n current_time = int(time.time())\n\n # Check if token needs refresh (5 minute buffer)\n if tokens['expires_at'] <= current_time + 300:\n print(\"\u26a0\ufe0f Token expired or expiring soon, refreshing...\")\n return self.refresh_and_get_client(tokens['refresh_token'])\n\n # Return client with valid token\n return SellerLegendClient(\n base_url=self.base_url,\n access_token=tokens['access_token']\n )\n\n def refresh_and_get_client(self, refresh_token):\n \"\"\"\n Refresh tokens and return authenticated client.\n\n \u26a0\ufe0f CRITICAL: When refreshing, you get a NEW refresh token!\n The old refresh token becomes INVALID immediately!\n \"\"\"\n client = SellerLegendClient(\n client_id=self.client_id,\n client_secret=self.client_secret,\n base_url=self.base_url\n )\n\n try:\n # Get new tokens\n new_tokens = client.refresh_access_token(refresh_token)\n\n # CRITICAL: Store the NEW tokens (including refresh token!)\n self.store_tokens(new_tokens)\n\n print(\"\u2705 Tokens refreshed successfully\")\n print(\"\u26a0\ufe0f WARNING: Old refresh token is now INVALID!\")\n\n # Return client with new access token\n return SellerLegendClient(\n base_url=self.base_url,\n access_token=new_tokens['access_token']\n )\n\n except Exception as e:\n print(f\"\u274c Token refresh failed: {e}\")\n print(\"User must re-authenticate with OAuth flow\")\n raise\n\n# Usage Example\nif __name__ == \"__main__\":\n # Initialize token manager\n manager = SellerLegendTokenManager(\n client_id=\"YOUR_CLIENT_ID\",\n client_secret=\"YOUR_CLIENT_SECRET\"\n )\n\n # First time: Complete OAuth flow\n # Uncomment this line for initial setup:\n # manager.complete_oauth_flow()\n\n # Subsequent uses: Get authenticated client\n try:\n client = manager.get_valid_client()\n\n # Now you can make API calls\n user = client.user.get_me()\n print(f\"Authenticated as: {user['user']['name']}\")\n\n # Get accounts\n accounts = client.user.get_accounts()\n for account in accounts['accounts']:\n print(f\"Account: {account['account_title']}\")\n\n # Get sales data\n orders = client.sales.get_orders(\n per_page=500,\n start_date=\"2024-01-01\",\n end_date=\"2024-01-31\"\n )\n print(f\"Found {len(orders.get('data', []))} orders\")\n\n except Exception as e:\n print(f\"Error: {e}\")\n print(\"Please run complete_oauth_flow() to authenticate\")\n```\n\n### 2. Token Refresh - Critical Information\n\n\u26a0\ufe0f **CRITICAL POINTS ABOUT TOKEN REFRESH**:\n\n1. **NEW Refresh Token**: When you refresh an access token, you receive a NEW refresh token\n2. **Old Token Invalid**: The old refresh token becomes invalid IMMEDIATELY\n3. **Must Update Storage**: You MUST update your stored refresh token every time\n4. **Atomic Updates**: Use database transactions to prevent token loss\n5. **Handle Failures**: If refresh fails, user must re-authenticate\n\n```python\ndef refresh_tokens_safely(old_refresh_token, client_id, client_secret):\n \"\"\"\n Safe token refresh with proper error handling.\n \"\"\"\n client = SellerLegendClient(\n client_id=client_id,\n client_secret=client_secret,\n base_url=\"https://app.sellerlegend.com\"\n )\n\n try:\n # Refresh tokens\n new_tokens = client.refresh_access_token(old_refresh_token)\n\n # CRITICAL: You now have NEW tokens\n # The old refresh token is INVALID!\n\n # Store BOTH new tokens immediately\n with database.transaction(): # Use transaction for safety\n database.execute(\"\"\"\n UPDATE oauth_tokens SET\n access_token = ?,\n refresh_token = ?, # NEW refresh token!\n expires_at = ?,\n updated_at = ?\n WHERE refresh_token = ?\n \"\"\", [\n new_tokens['access_token'],\n new_tokens['refresh_token'], # Must store the NEW one!\n time.time() + new_tokens['expires_in'],\n time.time(),\n old_refresh_token # Match on old token\n ])\n\n print(\"\u2705 Tokens refreshed and stored\")\n print(\"\u26a0\ufe0f Old refresh token is now invalid!\")\n\n return new_tokens\n\n except Exception as e:\n # Refresh failed - tokens may be invalid\n print(f\"\u274c Token refresh failed: {e}\")\n\n # Mark tokens as invalid in database\n database.execute(\n \"UPDATE oauth_tokens SET expires_at = 0 WHERE refresh_token = ?\",\n [old_refresh_token]\n )\n\n # User must re-authenticate\n raise Exception(\"Token refresh failed. User must re-authenticate.\")\n```\n\n### 3. Common Pitfalls and Solutions\n\n#### Pitfall 1: Not Storing Refresh Token\n```python\n# \u274c WRONG - Only storing access token\ndatabase.save(access_token=token_info['access_token'])\n\n# \u2705 CORRECT - Store both tokens\ndatabase.save(\n access_token=token_info['access_token'],\n refresh_token=token_info['refresh_token']\n)\n```\n\n#### Pitfall 2: Not Updating Refresh Token After Refresh\n```python\n# \u274c WRONG - Not updating refresh token\nnew_tokens = client.refresh_access_token(old_refresh_token)\ndatabase.update(access_token=new_tokens['access_token']) # Missing refresh token!\n\n# \u2705 CORRECT - Update both tokens\nnew_tokens = client.refresh_access_token(old_refresh_token)\ndatabase.update(\n access_token=new_tokens['access_token'],\n refresh_token=new_tokens['refresh_token'] # Must update this too!\n)\n```\n\n#### Pitfall 3: Using Expired Refresh Token\n```python\n# \u274c WRONG - Using old refresh token after refresh\ntokens1 = client.refresh_access_token(refresh_token)\n# ... later ...\ntokens2 = client.refresh_access_token(refresh_token) # Will fail!\n\n# \u2705 CORRECT - Use the new refresh token\ntokens1 = client.refresh_access_token(old_refresh_token)\n# ... later ...\ntokens2 = client.refresh_access_token(tokens1['refresh_token']) # Use new token\n```\n\n## API Usage Examples\n\nOnce you have a valid authenticated client, you can access all API endpoints:\n\n### Sales Data\n\n```python\n# Get authenticated client\nclient = token_manager.get_valid_client()\n\n# Get recent orders\norders = client.sales.get_orders(\n per_page=500,\n start_date=\"2023-12-01\",\n end_date=\"2023-12-31\"\n)\n\n# Get sales statistics\nstats = client.sales.get_statistics_dashboard(\n view_by=\"product\",\n group_by=\"sku\",\n per_page=1000,\n currency=\"USD\"\n)\n\n# Get daily sales per product\ndaily_sales = client.sales.get_per_day_per_product(\n per_page=500,\n start_date=\"2023-12-01\",\n end_date=\"2023-12-31\"\n)\n\n# Get transaction data\ntransactions = client.sales.get_transactions(\n per_page=1000,\n start_date=\"2023-12-01\",\n end_date=\"2023-12-31\"\n)\n```\n\n### Reports\n\n```python\n# Create and download a report (async)\nreport_data = client.reports.create_and_download_report(\n product_sku=\"YOUR-SKU-123\",\n timeout=300 # Wait up to 5 minutes\n)\n\n# Save the report to file\nwith open(\"report.csv.gz\", \"wb\") as f:\n f.write(report_data)\n\n# Or handle the process manually\n# 1. Create report request\ncreate_response = client.reports.create_report_request(\n product_sku=\"YOUR-SKU-123\"\n)\nreport_id = create_response[\"report_id\"]\n\n# 2. Check status periodically\nstatus = client.reports.get_report_status(report_id)\nprint(f\"Report status: {status['status']}\")\n\n# 3. Download when ready\nif status['status'] == 'done':\n report_data = client.reports.download_report(report_id)\n```\n\n### Inventory Management\n\n```python\n# Get inventory list\ninventory = client.inventory.get_list(\n per_page=500,\n velocity_start_date=\"2023-11-01\",\n velocity_end_date=\"2023-12-31\"\n)\n\n# Filter by specific SKU\ninventory_filtered = client.inventory.get_list(\n per_page=100,\n filter_by=\"sku\",\n filter_value=\"YOUR-SKU-123\"\n)\n```\n\n### Cost Management (COGS)\n\n```python\n# Get cost periods for a product\ncost_periods = client.costs.get_cost_periods(sku=\"YOUR-SKU-123\")\n\n# Update cost periods\ncost_data = [\n {\n \"dates\": {\n \"from_date\": \"2023-01-01\",\n \"to_date\": \"2023-12-31\"\n },\n \"cost_elements\": [\n {\n \"cost_element\": \"Product Cost\",\n \"provider\": \"Supplier Name\",\n \"notes\": \"Updated cost\",\n \"total_amount\": 15.50,\n \"currency\": \"USD\",\n \"conversion_rate\": \"1.00\",\n \"units\": 1,\n \"amount\": 15.50\n }\n ]\n }\n]\n\nresult = client.costs.update_cost_periods(\n sku=\"YOUR-SKU-123\",\n data=cost_data\n)\n```\n\n### Supply Chain\n\n```python\n# Get restock suggestions\nsuggestions = client.supply_chain.get_restock_suggestions(\n per_page=500,\n currency=\"USD\"\n)\n\nfor suggestion in suggestions['data']:\n print(f\"SKU: {suggestion['sku']}, Suggested Reorder: {suggestion['suggested_quantity']}\")\n```\n\n### Connections Status\n\n```python\n# Check Amazon connection status\nconnections = client.connections.get_list()\n\nfor conn in connections['connections']:\n print(f\"Account: {conn['account_title']}\")\n print(f\"SP API Status: {conn['sp']['status']}\")\n print(f\"PPC Status: {conn['ppc']['status']}\")\n```\n\n## Account Filtering\n\nMost endpoints support filtering by specific accounts or account groups:\n\n```python\n# Filter by account title\norders = client.sales.get_orders(\n account_title=\"My Store\",\n per_page=500\n)\n\n# Filter by seller ID and marketplace\norders = client.sales.get_orders(\n seller_id=\"A1SELLER123\",\n marketplace_id=\"ATVPDKIKX0DER\",\n per_page=500\n)\n\n# Filter by account group\nstats = client.sales.get_statistics_dashboard(\n view_by=\"product\",\n group_by=\"sku\",\n group_title=\"US Stores\",\n per_page=1000\n)\n```\n\n## Error Handling\n\n```python\nfrom sellerlegend_api import (\n SellerLegendClient,\n AuthenticationError,\n ValidationError,\n NotFoundError,\n RateLimitError,\n ServerError\n)\n\ntry:\n # Get authenticated client\n client = token_manager.get_valid_client()\n\n # Make API calls\n orders = client.sales.get_orders(per_page=500)\n\nexcept AuthenticationError as e:\n print(f\"Authentication failed: {e.message}\")\n # Token may be expired, try refresh or re-authenticate\n\nexcept ValidationError as e:\n print(f\"Validation error: {e.message}\")\n print(f\"Response data: {e.response_data}\")\n\nexcept NotFoundError as e:\n print(f\"Resource not found: {e.message}\")\n\nexcept RateLimitError as e:\n print(f\"Rate limit exceeded: {e.message}\")\n # Implement backoff strategy\n\nexcept ServerError as e:\n print(f\"Server error: {e.message}\")\n```\n\n## Configuration\n\n### Timeout and Retry Settings\n\n```python\nclient = SellerLegendClient(\n client_id=\"your_client_id\",\n client_secret=\"your_client_secret\",\n base_url=\"https://app.sellerlegend.com\",\n timeout=60, # 60 second timeout\n max_retries=5, # Retry up to 5 times\n backoff_factor=0.5 # Backoff factor for retries\n)\n```\n\n## Available Resources\n\nThe SDK provides access to the following SellerLegend API resources:\n\n- **`client.user`**: User information and account management\n- **`client.sales`**: Orders, statistics, and transaction data\n- **`client.reports`**: Report generation and download\n- **`client.inventory`**: Inventory levels and management\n- **`client.costs`**: Cost of goods sold (COGS) management\n- **`client.connections`**: Amazon connection status\n- **`client.supply_chain`**: Restock suggestions\n- **`client.warehouse`**: Warehouse and inbound shipment data\n- **`client.notifications`**: Notification management\n\n## Rate Limiting\n\nThe SDK includes built-in retry logic for rate-limited requests. When a rate limit is encountered (HTTP 429), the client will automatically retry with exponential backoff.\n\n## Important Security Notes\n\n1. **Never commit credentials**: Keep your `client_id` and `client_secret` secure\n2. **Use environment variables**: Store sensitive data in `.env` files\n3. **Secure token storage**: Always store tokens in a database, never in files or cookies\n4. **Use HTTPS**: Always use HTTPS for callbacks and API calls\n5. **Validate state parameter**: Always validate the state parameter to prevent CSRF attacks\n\n## Database Schema Example\n\n```sql\n-- Recommended schema for token storage\nCREATE TABLE oauth_tokens (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n user_id INTEGER, -- Your application's user ID\n access_token TEXT NOT NULL,\n refresh_token TEXT NOT NULL,\n expires_at INTEGER NOT NULL,\n created_at INTEGER NOT NULL,\n updated_at INTEGER,\n UNIQUE(access_token)\n);\n\n-- Index for faster lookups\nCREATE INDEX idx_user_id ON oauth_tokens(user_id);\nCREATE INDEX idx_expires_at ON oauth_tokens(expires_at);\n```\n\n## Troubleshooting\n\n### \"Unauthenticated\" Error\n- Token may be expired - check `expires_at` timestamp\n- Try refreshing the token using the refresh token\n- If refresh fails, user must re-authenticate\n\n### \"Access Denied\" Error\n- User may not have API access enabled\n- Check account permissions in SellerLegend\n\n### Token Refresh Fails\n- Refresh token may be expired (30 days)\n- Refresh token may have been used already (they're single-use)\n- User must complete OAuth flow again\n\n### Rate Limiting\n- Implement exponential backoff\n- Cache frequently accessed data\n- Use batch endpoints where available\n\n## Support\n\nFor support, please contact [support@sellerlegend.com](mailto:support@sellerlegend.com) or visit our [documentation](https://docs.sellerlegend.com/api-docs/index.html).\n\n## License\n\nThis SDK is licensed under the MIT License. See LICENSE file for details.\n",
"bugtrack_url": null,
"license": "MIT License\n \n Copyright (c) 2024 SellerLegend\n \n Permission is hereby granted, free of charge, to any person obtaining a copy\n of this software and associated documentation files (the \"Software\"), to deal\n in the Software without restriction, including without limitation the rights\n to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\n copies of the Software, and to permit persons to whom the Software is\n furnished to do so, subject to the following conditions:\n \n The above copyright notice and this permission notice shall be included in all\n copies or substantial portions of the Software.\n \n THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\n IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\n FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\n AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\n LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\n SOFTWARE.",
"summary": "Official Python SDK for the SellerLegend API",
"version": "1.0.3",
"project_urls": {
"Changelog": "https://github.com/sellerlegend/sellerlegend_api_py/blob/main/CHANGELOG.md",
"Documentation": "https://dashboard.sellerlegend.com/api-docs/index.html",
"Homepage": "https://github.com/sellerlegend/sellerlegend_api_py",
"Issues": "https://github.com/sellerlegend/sellerlegend_api_py/issues",
"Repository": "https://github.com/sellerlegend/sellerlegend_api_py"
},
"split_keywords": [
"sellerlegend",
" api",
" amazon",
" seller",
" ecommerce",
" analytics",
" amazon-seller",
" marketplace",
" fba",
" seller-central"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "aa3f46c8ebbcce5882b625335929642431e329d8deb88849a979c99617b391ea",
"md5": "a94bfa5b815172d308d574e4c2acf49d",
"sha256": "ca4320f40bbe61069348b1938a0c272d692e3e934b81bd22ea972b48f8629de6"
},
"downloads": -1,
"filename": "sellerlegend_api-1.0.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "a94bfa5b815172d308d574e4c2acf49d",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 30379,
"upload_time": "2025-09-19T05:31:52",
"upload_time_iso_8601": "2025-09-19T05:31:52.998502Z",
"url": "https://files.pythonhosted.org/packages/aa/3f/46c8ebbcce5882b625335929642431e329d8deb88849a979c99617b391ea/sellerlegend_api-1.0.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "941f4de7d9d5e53e8110fa264e87e84b6f735839c5de1c678518eabee36053c0",
"md5": "0bf7c0282c723100a11c82a95ba2ae38",
"sha256": "5427ddc53c77ff50ea7da824fb45e9daa0870fe2aff2b85e258e35e4b559d67f"
},
"downloads": -1,
"filename": "sellerlegend_api-1.0.3.tar.gz",
"has_sig": false,
"md5_digest": "0bf7c0282c723100a11c82a95ba2ae38",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 39489,
"upload_time": "2025-09-19T05:31:54",
"upload_time_iso_8601": "2025-09-19T05:31:54.747211Z",
"url": "https://files.pythonhosted.org/packages/94/1f/4de7d9d5e53e8110fa264e87e84b6f735839c5de1c678518eabee36053c0/sellerlegend_api-1.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-19 05:31:54",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "sellerlegend",
"github_project": "sellerlegend_api_py",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [
{
"name": "requests",
"specs": [
[
">=",
"2.28.0"
]
]
},
{
"name": "requests-oauthlib",
"specs": [
[
">=",
"1.3.0"
]
]
},
{
"name": "python-dateutil",
"specs": [
[
">=",
"2.8.0"
]
]
},
{
"name": "typing-extensions",
"specs": [
[
">=",
"4.0.0"
]
]
},
{
"name": "python-dotenv",
"specs": [
[
">=",
"1.0.0"
]
]
}
],
"lcname": "sellerlegend-api"
}
