如何高效管理aqicn token：从获取到安全调用的全链路实践

1次阅读

没有评论

共计 2716 个字符，预计需要花费 7 分钟才能阅读完成。

在使用 aqicn 空气质量数据 API 时，开发者常遇到以下几个典型问题：

Token 泄露风险 ：直接硬编码 token 在客户端或配置文件中，可能导致敏感信息泄露
调用频率限制 ：超过 API 规定的请求速率会导致服务被临时封禁
多环境管理困难 ：开发、测试、生产环境使用相同 token，难以隔离和追踪问题
自动续期缺失 ：token 过期后需人工干预，影响服务连续性

获取层 ：负责与 aqicn 认证服务交互，处理 token 的申请和刷新
缓存层 ：实现多级缓存策略，平衡响应速度和数据安全性
调度层 ：管理请求队列，实施流量控制和错误重试机制

内存缓存 （如 Redis/Memcached）：
优点：毫秒级响应，适合高频访问
缺点：服务重启后数据丢失
持久化存储 （如数据库 / 加密文件）：
优点：数据可持久保存
缺点：读写性能较低

推荐采用分层缓存：内存缓存作为一级缓存，持久化存储作为后备方案。

令牌桶算法示例（Python）：

from threading import Lock
import time

class TokenBucket:
    def __init__(self, capacity, fill_rate):
        self.capacity = float(capacity)
        self._tokens = float(capacity)
        self.fill_rate = float(fill_rate)
        self.timestamp = time.time()
        self.lock = Lock()

    def consume(self, tokens=1):
        with self.lock:
            self._add_new_tokens()
            if self._tokens >= tokens:
                self._tokens -= tokens
                return True
            return False

    def _add_new_tokens(self):
        now = time.time()
        elapsed = now - self.timestamp
        self.timestamp = now
        self._tokens = min(
            self.capacity, 
            self._tokens + elapsed * self.fill_rate
        )

import os
from cryptography.fernet import Fernet
import requests
from datetime import datetime, timedelta

class AQICNTokenManager:
    def __init__(self):
        self.cache_file = 'token_cache.enc'
        self._load_encryption_key()

    def _load_encryption_key(self):
        """从环境变量加载加密密钥"""
        key = os.getenv('AQICN_ENCRYPTION_KEY')
        if not key:
            raise ValueError('加密密钥未配置')
        self.cipher = Fernet(key.encode())

    def _fetch_new_token(self):
        """从 aqicn 获取新 token"""
        # 实际实现中替换为真实 API 调用
        mock_response = {'token': 'demo_token_' + datetime.now().strftime('%Y%m%d%H%M%S'),
            'expires_in': 3600
        }
        return {'token': mock_response['token'],
            'expiry': datetime.now() + timedelta(seconds=mock_response['expires_in'])
        }

    def _save_token(self, token_data):
        """加密存储 token 到文件"""
        data_str = f"{token_data['token']}|{token_data['expiry'].isoformat()}"
        encrypted = self.cipher.encrypt(data_str.encode())
        with open(self.cache_file, 'wb') as f:
            f.write(encrypted)

    def get_valid_token(self):
        """获取有效 token，自动处理续期"""
        # 尝试从缓存读取
        try:
            with open(self.cache_file, 'rb') as f:
                decrypted = self.cipher.decrypt(f.read()).decode()
                token, expiry_str = decrypted.split('|')
                expiry = datetime.fromisoformat(expiry_str)

                if datetime.now() < expiry - timedelta(minutes=5):
                    return token
        except (FileNotFoundError, ValueError):
            pass

        # 获取新 token 并缓存
        new_token = self._fetch_new_token()
        self._save_token(new_token)
        return new_token['token']