Python Client
简介
Client封装了请求签名、获取access_token、刷新access_token等功能。 通过Python Client可快速发起FastMoss API 请求。 其中本地存储token功能需调用方来实现。( _cache_token 和 _get_cached_token 函数)
request example
Python
from fastmoss_api_client import FastMossClient
client = FastMossClient(client_id="client_id", client_secret="client_secret")
response = client.do_api_call("/test", {"hello": "world"})Python Client Code:
Python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
fastmoss_api_client.py - A Python client for FastMoss API
This module provides a complete implementation for:
- Token management (acquisition and refresh)
- API request signing
- Local token caching
- Standardized API calls
#NOTE: _cache_token and _get_cached_token methods must be implemented.
Example usage:
>>> from fastmoss_api_client import FastMossClient
>>> client = FastMossClient(client_id="your_id", client_secret="your_secret")
>>> token = client.get_token()
>>> response = client.do_api_call("/v1/videos", {"page": 1})
"""
import hashlib
import json
import logging
import time
from typing import Any, Dict, Optional
import requests
from requests.exceptions import RequestException
# Set up logging
logger = logging.getLogger(__name__)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
class FastMossError(Exception):
"""Base exception for FastMoss API errors."""
pass
class FastMossClient:
"""A client for interacting with FastMoss API."""
def __init__(
self,
client_id: str,
client_secret: str,
base_url: str = "https://openapi.test.fastmoss.com",
timeout: int = 15
):
"""Initialize the FastMoss API client.
Args:
client_id: API client identifier
client_secret: API client secret key
base_url: Base API URL (defaults to test environment)
timeout: Request timeout in seconds
"""
self.client_id = client_id
self.client_secret = client_secret
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.session = requests.Session()
def _get_new_token(self) -> Dict[str, Any]:
"""Request a new access token from FastMoss API.
Returns:
Dictionary containing token information or error
Raises:
FastMossError: If token request fails
"""
url = f"{self.base_url}/v1/token"
headers = {"Content-Type": "application/json"}
payload = {
"client_id": self.client_id,
"client_secret": self.client_secret
}
try:
response = self.session.post(
url,
headers=headers,
json=payload,
timeout=self.timeout
)
response.raise_for_status()
response_data = response.json()
if response_data.get("code") == 0:
return response_data.get("data", {})
else:
error_msg = response_data.get("message", "Unknown error")
logger.error(f"Token request failed: {error_msg}")
raise FastMossError(error_msg)
except RequestException as e:
logger.error(f"Token request exception: {str(e)}")
raise FastMossError(f"Token request failed: {str(e)}")
def _refresh_token(self, refresh_token: str) -> Dict[str, Any]:
"""Refresh an expired access token.
Args:
refresh_token: The refresh token from previous authentication
Returns:
Dictionary containing new token information or error
Raises:
FastMossError: If token refresh fails
"""
url = f"{self.base_url}/v1/refreshToken"
headers = {"Content-Type": "application/json"}
payload = {
"client_id": self.client_id,
"refresh_token": refresh_token
}
try:
response = self.session.post(
url,
headers=headers,
json=payload,
timeout=self.timeout
)
response.raise_for_status()
response_data = response.json()
if response_data.get("code") == 0:
return response_data.get("data", {})
else:
error_msg = response_data.get("message", "Unknown error")
logger.error(f"Token refresh failed: {error_msg}")
raise FastMossError(error_msg)
except RequestException as e:
logger.error(f"Token refresh exception: {str(e)}")
raise FastMossError(f"Token refresh failed: {str(e)}")
def _get_cached_token(self) -> Optional[Dict[str, Any]]:
"""Retrieve cached token from local storage.
Note:
This is a placeholder implementation. In production, you would
implement actual storage/retrieval from Redis, database, etc.
Returns:
Cached token dictionary if available, None otherwise
"""
# TODO: Implement actual storage/retrieval logic
# Example: token_data = cache.get(f"fm_token:{self.client_id}")
return None
def _cache_token(self, token_data: Dict[str, Any]) -> bool:
"""Cache token data in local storage.
Note:
This is a placeholder implementation. In production, you would
implement actual storage in Redis, database, etc.
Args:
token_data: Token information to cache
Returns:
True if caching succeeded, False otherwise
"""
# TODO: Implement actual storage logic
# Example: cache.set(f"fm_token:{self.client_id}", json.dumps(token_data, ensure_ascii=False),token_data.get("refresh_expires_in", 86400))
return True
def _is_token_valid(self, token_data: Dict[str, Any]) -> bool:
"""Check if token is still valid.
Args:
token_data: Token information dictionary
Returns:
True if token is valid, False otherwise
"""
expire_at = token_data.get("expire_at", 0)
current_time = time.time()
return expire_at > current_time + 600 # 10 minute buffer
def _is_refresh_token_valid(self, token_data: Dict[str, Any]) -> bool:
"""Check if refresh token is still valid.
Args:
token_data: Token information dictionary
Returns:
True if refresh token is valid, False otherwise
"""
refresh_expire_at = token_data.get("refresh_expire_at", 0)
current_time = time.time()
return refresh_expire_at > current_time + 600 # 10 minute buffer
def get_token(self) -> Dict[str, Any]:
"""Get a valid access token, using cache if available.
Returns:
Dictionary containing valid token information
Raises:
FastMossError: If token acquisition fails
"""
# Try to get cached token
token_data = self._get_cached_token()
if token_data:
if self._is_token_valid(token_data):
return token_data
elif self._is_refresh_token_valid(token_data):
try:
new_token = self._refresh_token(token_data["refresh_token"])
self._cache_token(new_token)
return new_token
except FastMossError:
logger.warning("Token refresh failed, getting new token")
# Get new token if cache is empty or refresh failed
try:
new_token = self._get_new_token()
self._cache_token(new_token)
return new_token
except FastMossError as e:
logger.error("Failed to get new token")
raise FastMossError("Failed to acquire valid token") from e
def _generate_signature(self, uri: str, json_str: str) -> str:
"""Generate API request signature.
Args:
uri: API endpoint URI
json_str: JSON string of request payload
Returns:
SHA256 signature string
"""
sign_data = f"{self.client_secret}|{uri}|{json_str}|{self.client_secret}"
print("sign_data",sign_data)
return hashlib.sha256(sign_data.encode('utf-8')).hexdigest()
def do_api_call(
self,
uri: str,
post_data: Dict[str, Any],
method: str = "POST"
) -> Dict[str, Any]:
"""Make an authenticated API call to FastMoss.
Args:
uri: API endpoint URI
post_data: Dictionary of request parameters
method: HTTP method (default: POST)
Returns:
Dictionary containing API response data
Raises:
FastMossError: If API request fails
"""
try:
# Get valid token
token_info = self.get_token()
access_token = token_info.get("access_token")
if not access_token:
raise FastMossError("Access token not available")
# Prepare request
post_data_str = json.dumps(post_data, ensure_ascii=False)
signature = self._generate_signature(uri, post_data_str)
timestamp = int(time.time())
url = (
f"{self.base_url}{uri}?"
f"access_token={access_token}&"
f"sign={signature}&"
f"client_id={self.client_id}&"
f"timestamp={timestamp}&"
f"signature_version=2"
)
headers = {
'Content-Type': 'application/json; charset=utf-8',
'Content-Length': str(len(post_data_str))
}
# Make request
response = self.session.request(
method,
url,
data=post_data_str.encode('utf-8'),
headers=headers,
timeout=self.timeout
)
response.raise_for_status()
return response.json()
except RequestException as e:
error_msg = f"API request failed: {str(e)}"
logger.error(error_msg)
raise FastMossError(error_msg) from e
except json.JSONDecodeError as e:
error_msg = f"Failed to decode API response: {str(e)}"
logger.error(error_msg)
raise FastMossError(error_msg) from e
def test(self, params: Dict[str, Any]) -> Dict[str, Any]:
""" FastMoss Test API.
Args:
params: Dictionary of query parameters
Returns:
Dictionary containing video list data
Raises:
FastMossError: If request fails
"""
try:
response = self.do_api_call("/test", params)
logger.info("Successfully call test")
return response
except FastMossError as e:
logger.error(f"Failed to call test: {str(e)}")
raise
# Example usage
if __name__ == "__main__":
try:
# Initialize client
client = FastMossClient(
client_id="your_client_id",
client_secret="your_client_secret"
)
#NOTE: _cache_token and _get_cached_token methods must be implemented.
# API call example
test_result = client.test({"page": 1, "limit": 10})
print("test_result:", test_result)
except FastMossError as e:
print(f"FastMoss Error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")