RESTful API客户端Development

"""
生成一个Python客户端, 用于调用GitHub API
要求: 
1. usingrequestslibrary
2. supportGET, POST, PUT, DELETEmethod
3. supportbasicauthentication and OAuthauthentication
4. package含errorprocessing
5. support速率限制processing
6. providing获取userinformation, 仓libraryinformation method
"""

import requests
import json
from typing import Dict, Optional, List, Any

class GitHubAPI:
    """
    GitHub API客户端
    """
    
    def __init__(self, token: Optional[str] = None, username: Optional[str] = None, password: Optional[str] = None):
        """
        初始化GitHub API客户端
        
        :param token: GitHub APItoken
        :param username: GitHubuser名 (用于basicauthentication) 
        :param password: GitHubpassword (用于basicauthentication) 
        """
        self.base_url = "https://api.github.com"
        self.session = requests.Session()
        
        # 设置authentication
        if token:
            self.session.headers.update({"Authorization": f"token {token}"})
        elif username and password:
            self.session.auth = (username, password)
        
        # 设置默认头部
        self.session.headers.update({
            "Accept": "application/vnd.github.v3+json",
            "User-Agent": "GitHub-API-Client"
        })
    
    def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        """
        发送APIrequest
        
        :param method: HTTPmethod
        :param endpoint: API端点
        :param kwargs: otherrequestparameter
        :return: responsedata
        :raises Exception: such as果request失败
        """
        url = f"{self.base_url}{endpoint}"
        
        try:
            response = self.session.request(method, url, **kwargs)
            
            # processing速率限制
            if response.status_code == 403 and "rate limit" in response.text.lower():
                reset_time = int(response.headers.get("X-RateLimit-Reset", 0))
                raise Exception(f"Rate limit exceeded. Try again at {reset_time}")
            
            # checkresponsestatus码
            response.raise_for_status()
            
            # processing空response
            if not response.content:
                return {}
            
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"API request failed: {str(e)}")
    
    def get_user(self, username: str) -> Dict[str, Any]:
        """
        获取userinformation
        
        :param username: GitHubuser名
        :return: userinformation
        """
        return self._request("GET", f"/users/{username}")
    
    def get_repositories(self, username: str, per_page: int = 30, page: int = 1) -> List[Dict[str, Any]]:
        """
        获取user仓librarylist
        
        :param username: GitHubuser名
        :param per_page: 每页数量
        :param page: 页码
        :return: 仓librarylist
        """
        return self._request("GET", f"/users/{username}/repos", params={"per_page": per_page, "page": page})
    
    def get_repository(self, owner: str, repo: str) -> Dict[str, Any]:
        """
        获取仓library详情
        
        :param owner: 仓library所 has 者
        :param repo: 仓library名称
        :return: 仓library详情
        """
        return self._request("GET", f"/repos/{owner}/{repo}")
    
    def create_repository(self, name: str, description: Optional[str] = None, private: bool = False) -> Dict[str, Any]:
        """
        creation new 仓library
        
        :param name: 仓library名称
        :param description: 仓librarydescribes
        :param private:  is 否私 has 
        :return: 仓libraryinformation
        """
        data = {
            "name": name,
            "description": description,
            "private": private
        }
        return self._request("POST", "/user/repos", json=data)
    
    def update_repository(self, owner: str, repo: str, **kwargs) -> Dict[str, Any]:
        """
        update仓libraryinformation
        
        :param owner: 仓library所 has 者
        :param repo: 仓library名称
        :param kwargs: updateparameter
        :return: update after  仓libraryinformation
        """
        return self._request("PATCH", f"/repos/{owner}/{repo}", json=kwargs)
    
    def delete_repository(self, owner: str, repo: str) -> Dict[str, Any]:
        """
        delete仓library
        
        :param owner: 仓library所 has 者
        :param repo: 仓library名称
        :return: responseinformation
        """
        return self._request("DELETE", f"/repos/{owner}/{repo}")

# example用法
if __name__ == "__main__":
    # 初始化客户端
    # usingtokenauthentication
    # api = GitHubAPI(token="your_github_token")
    
    # usingbasicauthentication
    # api = GitHubAPI(username="your_username", password="your_password")
    
    # 不usingauthentication (速率限制较严格) 
    api = GitHubAPI()
    
    # 获取userinformation
    try:
        user = api.get_user("octocat")
        print(f"User: {user['login']}")
        print(f"Name: {user.get('name', 'N/A')}")
        print(f"Bio: {user.get('bio', 'N/A')}")
        print(f"Public repos: {user.get('public_repos', 0)}")
        
        # 获取user仓library
        repos = api.get_repositories("octocat", per_page=5)
        print("\nRepositories:")
        for repo in repos:
            print(f"- {repo['name']}: {repo.get('description', 'No description')}")
    except Exception as e:
        print(f"Error: {str(e)}")

"""
生成一个JavaScript客户端, 用于调用RESTful API
要求: 
1. usingfetch API
2. supportGET, POST, PUT, DELETEmethod
3. supportbasicauthentication and Bearertokenauthentication
4. package含errorprocessing
5. supportJSONdata格式
"""

class APIClient {
    /**
     * 初始化API客户端
     * @param {string} baseURL - APIBasicsURL
     * @param {Object} options - configuration选项
     */
    constructor(baseURL, options = {}) {
        this.baseURL = baseURL;
        this.headers = options.headers || {
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        };
        
        // 设置authentication
        if (options.token) {
            this.headers['Authorization'] = `Bearer ${options.token}`;
        } else if (options.username && options.password) {
            const credentials = btoa(`${options.username}:${options.password}`);
            this.headers['Authorization'] = `Basic ${credentials}`;
        }
    }
    
    /**
     * 发送APIrequest
     * @param {string} endpoint - API端点
     * @param {string} method - HTTPmethod
     * @param {Object} data - requestdata
     * @returns {Promise} - responsedata
     */
    async request(endpoint, method = 'GET', data = null) {
        const url = `${this.baseURL}${endpoint}`;
        
        const config = {
            method,
            headers: this.headers,
            credentials: 'include'
        };
        
        if (data && (method === 'POST' || method === 'PUT' || method === 'PATCH')) {
            config.body = JSON.stringify(data);
        }
        
        try {
            const response = await fetch(url, config);
            
            // checkresponsestatus
            if (!response.ok) {
                const errorData = await response.json().catch(() => ({}));
                throw new Error(errorData.message || `API error: ${response.status}`);
            }
            
            // processing空response
            if (response.status === 204) {
                return {};
            }
            
            return await response.json();
        } catch (error) {
            throw new Error(`Request failed: ${error.message}`);
        }
    }
    
    /**
     * 发送GETrequest
     * @param {string} endpoint - API端点
     * @param {Object} params - queryparameter
     * @returns {Promise} - responsedata
     */
    async get(endpoint, params = {}) {
        // 构建querystring
        const queryString = new URLSearchParams(params).toString();
        const url = queryString ? `${endpoint}?${queryString}` : endpoint;
        return this.request(url, 'GET');
    }
    
    /**
     * 发送POSTrequest
     * @param {string} endpoint - API端点
     * @param {Object} data - requestdata
     * @returns {Promise} - responsedata
     */
    async post(endpoint, data) {
        return this.request(endpoint, 'POST', data);
    }
    
    /**
     * 发送PUTrequest
     * @param {string} endpoint - API端点
     * @param {Object} data - requestdata
     * @returns {Promise} - responsedata
     */
    async put(endpoint, data) {
        return this.request(endpoint, 'PUT', data);
    }
    
    /**
     * 发送DELETErequest
     * @param {string} endpoint - API端点
     * @returns {Promise} - responsedata
     */
    async delete(endpoint) {
        return this.request(endpoint, 'DELETE');
    }
    
    /**
     * 发送PATCHrequest
     * @param {string} endpoint - API端点
     * @param {Object} data - requestdata
     * @returns {Promise} - responsedata
     */
    async patch(endpoint, data) {
        return this.request(endpoint, 'PATCH', data);
    }
}

// example用法
async function main() {
    // 初始化客户端
    const api = new APIClient('https://api.example.com', {
        // token: 'your-token-here'
        // username: 'your-username',
        // password: 'your-password'
    });
    
    try {
        // 获取data
        const users = await api.get('/users', { page: 1, limit: 10 });
        console.log('Users:', users);
        
        // creationdata
        const newUser = await api.post('/users', {
            name: 'John Doe',
            email: 'john@example.com'
        });
        console.log('Created user:', newUser);
        
        // updatedata
        const updatedUser = await api.put(`/users/${newUser.id}`, {
            name: 'John Smith'
        });
        console.log('Updated user:', updatedUser);
        
        // deletedata
        await api.delete(`/users/${newUser.id}`);
        console.log('User deleted');
    } catch (error) {
        console.error('Error:', error.message);
    }
}

// runexample
if (typeof window === 'undefined') {
    // Node.jsenvironment
    main();
} else {
    // 浏览器environment
    window.APIClient = APIClient;
    console.log('APIClient is available in the global scope');
}

"""
生成一个Python客户端, implementationJWTauthentication
要求: 
1. usingrequestslibrary
2. supportJWTtoken获取 and 刷 new 
3. 自动processingtoken过期
4. package含errorprocessing
"""

import requests
import json
from typing import Dict, Optional, Any

class JWTAPIClient:
    """
    usingJWTauthentication API客户端
    """
    
    def __init__(self, base_url, auth_endpoint, client_id, client_secret):
        """
        初始化JWT API客户端
        
        :param base_url: APIBasicsURL
        :param auth_endpoint: authentication端点
        :param client_id: 客户端ID
        :param client_secret: 客户端key
        """
        self.base_url = base_url
        self.auth_endpoint = auth_endpoint
        self.client_id = client_id
        self.client_secret = client_secret
        self.token = None
        self.token_expiry = None
        
        self.session = requests.Session()
        self.session.headers.update({
            "Content-Type": "application/json",
            "Accept": "application/json"
        })
    
    def _get_token(self):
        """
        获取JWTtoken
        """
        url = f"{self.base_url}{self.auth_endpoint}"
        data = {
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "grant_type": "client_credentials"
        }
        
        response = self.session.post(url, json=data)
        response.raise_for_status()
        
        token_data = response.json()
        self.token = token_data.get("access_token")
        self.token_expiry = token_data.get("expires_in")
        
        # update头部
        self.session.headers.update({
            "Authorization": f"Bearer {self.token}"
        })
    
    def _request(self, method, endpoint, **kwargs):
        """
        发送APIrequest
        
        :param method: HTTPmethod
        :param endpoint: API端点
        :param kwargs: otherrequestparameter
        :return: responsedata
        """
        url = f"{self.base_url}{endpoint}"
        
        # 确保 has  has 效 token
        if not self.token:
            self._get_token()
        
        try:
            response = self.session.request(method, url, **kwargs)
            
            # processingtoken过期
            if response.status_code == 401:
                # 刷 new token
                self._get_token()
                # 重 new 发送request
                response = self.session.request(method, url, **kwargs)
            
            response.raise_for_status()
            
            if not response.content:
                return {}
            
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"API request failed: {str(e)}")
    
    def get(self, endpoint, params=None):
        """
        发送GETrequest
        """
        return self._request("GET", endpoint, params=params)
    
    def post(self, endpoint, data=None):
        """
        发送POSTrequest
        """
        return self._request("POST", endpoint, json=data)
    
    def put(self, endpoint, data=None):
        """
        发送PUTrequest
        """
        return self._request("PUT", endpoint, json=data)
    
    def delete(self, endpoint):
        """
        发送DELETErequest
        """
        return self._request("DELETE", endpoint)

# example用法
if __name__ == "__main__":
    # 初始化客户端
    client = JWTAPIClient(
        "https://api.example.com",
        "/auth/token",
        "your-client-id",
        "your-client-secret"
    )
    
    try:
        # 发送request
        response = client.get("/protected/resource")
        print("Response:", response)
        
        # 发送POSTrequest
        new_resource = client.post("/resources", {
            "name": "Test Resource",
            "description": "A test resource"
        })
        print("Created resource:", new_resource)
    except Exception as e:
        print(f"Error: {str(e)}")

"""
生成一个带 has 重试mechanism Python API客户端
要求: 
1. usingrequestslibrary
2. support自动重试
3. 可configuration重试次数 and latency
4. 只 for specificerrorfor重试
"""

import requests
import time
from typing import Dict, Optional, Any, List

class RetryAPIClient:
    """
    带 has 重试mechanism API客户端
    """
    
    def __init__(self, base_url, retries=3, retry_delay=1, retry_codes=None):
        """
        初始化API客户端
        
        :param base_url: APIBasicsURL
        :param retries: 重试次数
        :param retry_delay: 重试latency (秒) 
        :param retry_codes: 需要重试 status码
        """
        self.base_url = base_url
        self.retries = retries
        self.retry_delay = retry_delay
        self.retry_codes = retry_codes or [429, 500, 502, 503, 504]
        
        self.session = requests.Session()
        self.session.headers.update({
            "Content-Type": "application/json",
            "Accept": "application/json"
        })
    
    def _request(self, method, endpoint, **kwargs):
        """
        发送APIrequest, support重试
        
        :param method: HTTPmethod
        :param endpoint: API端点
        :param kwargs: otherrequestparameter
        :return: responsedata
        """
        url = f"{self.base_url}{endpoint}"
        
        last_error = None
        
        for attempt in range(self.retries + 1):
            try:
                response = self.session.request(method, url, **kwargs)
                
                # check is 否需要重试
                if response.status_code in self.retry_codes and attempt < self.retries:
                    last_error = Exception(f"Status code {response.status_code}")
                    print(f"Attempt {attempt + 1} failed, retrying in {self.retry_delay} seconds...")
                    time.sleep(self.retry_delay * (2 ** attempt))  # 指数退避
                    continue
                
                # checkresponsestatus码
                response.raise_for_status()
                
                # processing空response
                if not response.content:
                    return {}
                
                return response.json()
            except requests.exceptions.RequestException as e:
                # 只 for specificexceptionfor重试
                if isinstance(e, (requests.exceptions.ConnectionError, requests.exceptions.Timeout)) and attempt < self.retries:
                    last_error = e
                    print(f"Attempt {attempt + 1} failed, retrying in {self.retry_delay} seconds...")
                    time.sleep(self.retry_delay * (2 ** attempt))  # 指数退避
                    continue
                raise
        
        # 所 has 重试都失败
        if last_error:
            raise last_error
        
        raise Exception("Request failed with unknown error")
    
    def get(self, endpoint, params=None):
        """
        发送GETrequest
        """
        return self._request("GET", endpoint, params=params)
    
    def post(self, endpoint, data=None):
        """
        发送POSTrequest
        """
        return self._request("POST", endpoint, json=data)
    
    def put(self, endpoint, data=None):
        """
        发送PUTrequest
        """
        return self._request("PUT", endpoint, json=data)
    
    def delete(self, endpoint):
        """
        发送DELETErequest
        """
        return self._request("DELETE", endpoint)

# example用法
if __name__ == "__main__":
    # 初始化客户端
    client = RetryAPIClient(
        "https://api.example.com",
        retries=3,
        retry_delay=1
    )
    
    try:
        # 发送request
        response = client.get("/api/resources")
        print("Response:", response)
    except Exception as e:
        print(f"Error: {str(e)}")

"""
生成一个Python客户端, 用于调用OpenWeatherMap API
要求: 
1. usingrequestslibrary
2. support获取当 before 天气 and 天气预报
3. support基于城市名称 and 经纬度query
4. package含errorprocessing
5. support不同单位system
"""

import requests
from typing import Dict, Optional, Any

class WeatherAPIClient:
    """
    OpenWeatherMap API客户端
    """
    
    def __init__(self, api_key, base_url="https://api.openweathermap.org/data/2.5"):
        """
        初始化天气API客户端
        
        :param api_key: OpenWeatherMap APIkey
        :param base_url: APIBasicsURL
        """
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
    
    def _request(self, endpoint, params=None):
        """
        发送APIrequest
        
        :param endpoint: API端点
        :param params: queryparameter
        :return: responsedata
        """
        url = f"{self.base_url}{endpoint}"
        
        # 添加APIkey
        if params is None:
            params = {}
        params["appid"] = self.api_key
        
        try:
            response = self.session.get(url, params=params)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"API request failed: {str(e)}")
    
    def get_current_weather_by_city(self, city, units="metric"):
        """
        根据城市名称获取当 before 天气
        
        :param city: 城市名称
        :param units: 单位system, metric (摄氏度)  or  imperial (华氏度) 
        :return: 当 before 天气data
        """
        return self._request("/weather", {
            "q": city,
            "units": units
        })
    
    def get_current_weather_by_coords(self, lat, lon, units="metric"):
        """
        根据经纬度获取当 before 天气
        
        :param lat: 纬度
        :param lon: 经度
        :param units: 单位system
        :return: 当 before 天气data
        """
        return self._request("/weather", {
            "lat": lat,
            "lon": lon,
            "units": units
        })
    
    def get_forecast_by_city(self, city, units="metric", cnt=5):
        """
        根据城市名称获取天气预报
        
        :param city: 城市名称
        :param units: 单位system
        :param cnt: 预报天数
        :return: 天气预报data
        """
        return self._request("/forecast", {
            "q": city,
            "units": units,
            "cnt": cnt
        })
    
    def get_forecast_by_coords(self, lat, lon, units="metric", cnt=5):
        """
        根据经纬度获取天气预报
        
        :param lat: 纬度
        :param lon: 经度
        :param units: 单位system
        :param cnt: 预报天数
        :return: 天气预报data
        """
        return self._request("/forecast", {
            "lat": lat,
            "lon": lon,
            "units": units,
            "cnt": cnt
        })

# example用法
if __name__ == "__main__":
    # 初始化客户端
    api_key = "your_api_key_here"
    client = WeatherAPIClient(api_key)
    
    try:
        # 获取当 before 天气
        weather = client.get_current_weather_by_city("Beijing")
        print(f"Current weather in Beijing:")
        print(f"Temperature: {weather['main']['temp']}°C")
        print(f"Weather: {weather['weather'][0]['description']}")
        print(f"Humidity: {weather['main']['humidity']}%")
        print(f"Wind speed: {weather['wind']['speed']} m/s")
        
        # 获取天气预报
        forecast = client.get_forecast_by_city("Beijing", cnt=3)
        print("\n3-day forecast:")
        for item in forecast['list']:
            date = item['dt_txt']
            temp = item['main']['temp']
            desc = item['weather'][0]['description']
            print(f"{date}: {temp}°C, {desc}")
    except Exception as e:
        print(f"Error: {str(e)}")