qinglong API客户端：Python/Node.js调用示例

【免费下载链接】qinglong 支持 Python3、JavaScript、Shell、Typescript 的定时任务管理平台（Timed task management platform supporting Python3, JavaScript, Shell, Typescript） 项目地址: https://gitcode.com/GitHub_Trending/qi/qinglong

还在手动管理定时任务脚本？想要自动化集成青龙面板到你的DevOps流程中？本文将为你详细介绍如何通过Python和Node.js客户端高效调用qinglong API，实现定时任务的自动化管理。

读完本文，你将掌握：

• ✅ qinglong API的核心接口和认证机制
• ✅ Python客户端完整封装和调用示例
• ✅ Node.js客户端完整封装和调用示例
• ✅ 定时任务、环境变量、脚本管理的自动化操作
• ✅ 错误处理和最佳实践指南

qinglong API概览

qinglong提供了丰富的RESTful API接口，涵盖定时任务管理、环境变量配置、脚本操作等核心功能。所有API都遵循统一的响应格式：

{
  "code": 200,
  "data": {...},
  "message": "操作成功"
}

核心API端点

功能模块	端点路径	主要操作
定时任务	/crons	创建、查询、运行、停止任务
环境变量	/envs	增删改查环境变量
脚本管理	/scripts	文件上传、下载、执行
系统状态	/system	系统信息和健康检查
用户管理	/user	用户登录和权限管理

认证机制

qinglong使用Bearer Token进行API认证，需要在请求头中携带：

Authorization: Bearer <your_token>

Python客户端实现

安装依赖

pip install requests python-dotenv

基础客户端类

import requests
import json
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
import os
from dotenv import load_dotenv

@dataclass
class QinglongConfig:
    base_url: str = "http://localhost:5700"
    token: str = ""
    timeout: int = 30

class QinglongClient:
    def __init__(self, config: QinglongConfig = None):
        self.config = config or QinglongConfig()
        self.session = requests.Session()
        self._setup_session()
    
    def _setup_session(self):
        """设置会话头信息"""
        headers = {
            'Content-Type': 'application/json',
            'User-Agent': 'Qinglong-Python-Client/1.0.0'
        }
        if self.config.token:
            headers['Authorization'] = f'Bearer {self.config.token}'
        
        self.session.headers.update(headers)
    
    def _request(self, method: str, endpoint: str, **kwargs) -> Dict:
        """发送HTTP请求"""
        url = f"{self.config.base_url.rstrip('/')}{endpoint}"
        
        try:
            response = self.session.request(
                method, 
                url, 
                timeout=self.config.timeout,
                **kwargs
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise QinglongException(f"API请求失败: {str(e)}")
    
    def login(self, username: str, password: str) -> str:
        """用户登录获取token"""
        endpoint = "/user/login"
        data = {
            "username": username,
            "password": password
        }
        
        result = self._request('POST', endpoint, json=data)
        if result.get('code') == 200:
            self.config.token = result['data']['token']
            self._setup_session()  # 更新会话token
            return self.config.token
        else:
            raise QinglongException(f"登录失败: {result.get('message')}")
    
    # 定时任务管理
    def list_crons(self, search: str = None) -> List[Dict]:
        """获取定时任务列表"""
        params = {}
        if search:
            params['searchValue'] = search
        
        result = self._request('GET', '/crons', params=params)
        return result.get('data', [])
    
    def create_cron(self, task_data: Dict) -> Dict:
        """创建定时任务"""
        required_fields = ['name', 'command', 'schedule']
        for field in required_fields:
            if field not in task_data:
                raise QinglongException(f"缺少必要字段: {field}")
        
        result = self._request('POST', '/crons', json=task_data)
        return result.get('data', {})
    
    def run_cron(self, task_ids: List[int]) -> Dict:
        """运行定时任务"""
        result = self._request('PUT', '/crons/run', json=task_ids)
        return result.get('data', {})
    
    def stop_cron(self, task_ids: List[int]) -> Dict:
        """停止定时任务"""
        result = self._request('PUT', '/crons/stop', json=task_ids)
        return result.get('data', {})
    
    def delete_cron(self, task_ids: List[int]) -> Dict:
        """删除定时任务"""
        result = self._request('DELETE', '/crons', json=task_ids)
        return result.get('data', {})
    
    # 环境变量管理
    def list_envs(self, search: str = None) -> List[Dict]:
        """获取环境变量列表"""
        params = {}
        if search:
            params['searchValue'] = search
        
        result = self._request('GET', '/envs', params=params)
        return result.get('data', [])
    
    def create_env(self, env_data: Dict) -> Dict:
        """创建环境变量"""
        result = self._request('POST', '/envs', json=[env_data])
        return result.get('data', [{}])[0]
    
    def update_env(self, env_id: int, env_data: Dict) -> Dict:
        """更新环境变量"""
        env_data['id'] = env_id
        result = self._request('PUT', '/envs', json=env_data)
        return result.get('data', {})
    
    def delete_env(self, env_ids: List[int]) -> Dict:
        """删除环境变量"""
        result = self._request('DELETE', '/envs', json=env_ids)
        return result.get('data', {})
    
    # 脚本管理
    def list_scripts(self, path: str = None) -> List[Dict]:
        """获取脚本列表"""
        params = {}
        if path:
            params['path'] = path
        
        result = self._request('GET', '/scripts', params=params)
        return result.get('data', [])
    
    def get_script_content(self, file_path: str, file_name: str) -> str:
        """获取脚本内容"""
        params = {
            'path': file_path,
            'file': file_name
        }
        result = self._request('GET', '/scripts/detail', params=params)
        return result.get('data', '')
    
    def run_script(self, file_path: str, file_name: str, content: str = None) -> Dict:
        """运行脚本"""
        data = {
            'filename': file_name,
            'path': file_path,
            'content': content or ''
        }
        result = self._request('PUT', '/scripts/run', json=data)
        return result.get('data', {})

class QinglongException(Exception):
    """qinglong客户端异常"""
    pass

使用示例

# 配置客户端
from qinglong_client import QinglongClient, QinglongConfig

# 方式1: 使用环境变量配置
import os
from dotenv import load_dotenv

load_dotenv()

config = QinglongConfig(
    base_url=os.getenv('QL_URL', 'http://localhost:5700'),
    token=os.getenv('QL_TOKEN')
)

client = QinglongClient(config)

# 方式2: 用户名密码登录
if not config.token:
    client.login(
        username=os.getenv('QL_USERNAME', 'admin'),
        password=os.getenv('QL_PASSWORD', 'admin')
    )

# 定时任务管理示例
def manage_cron_tasks():
    # 获取所有定时任务
    tasks = client.list_crons()
    print(f"当前有 {len(tasks)} 个定时任务")
    
    # 创建新任务
    new_task = client.create_cron({
        "name": "每日数据备份",
        "command": "python3 /scripts/backup.py",
        "schedule": "0 2 * * *",
        "labels": ["backup", "daily"]
    })
    print(f"创建任务成功: {new_task['id']}")
    
    # 运行特定任务
    client.run_cron([new_task['id']])
    
    # 查询任务日志
    # 需要先获取任务详情，然后使用任务ID查询日志

# 环境变量管理示例
def manage_environment_variables():
    # 获取所有环境变量
    envs = client.list_envs()
    print(f"当前有 {len(envs)} 个环境变量")
    
    # 创建数据库连接环境变量
    db_env = client.create_env({
        "name": "DB_CONNECTION",
        "value": "mysql://user:pass@localhost:3306/db",
        "remarks": "生产环境数据库连接"
    })
    
    # 更新环境变量
    client.update_env(db_env['id'], {
        "name": "DB_CONNECTION",
        "value": "mysql://user:newpass@localhost:3306/db",
        "remarks": "更新后的数据库连接"
    })

# 脚本管理示例
def manage_scripts():
    # 获取脚本列表
    scripts = client.list_scripts()
    
    # 读取脚本内容
    content = client.get_script_content('', 'backup.py')
    print(f"备份脚本内容长度: {len(content)} 字符")
    
    # 运行脚本
    result = client.run_script('', 'backup.py')
    print(f"脚本执行结果: {result}")

if __name__ == "__main__":
    try:
        manage_cron_tasks()
        manage_environment_variables()
        manage_scripts()
    except QinglongException as e:
        print(f"操作失败: {e}")

Node.js客户端实现

安装依赖

npm install axios dotenv

基础客户端类

const axios = require('axios');
const fs = require('fs');
const path = require('path');

class QinglongConfig {
    constructor(baseUrl = 'http://localhost:5700', token = '', timeout = 30000) {
        this.baseUrl = baseUrl;
        this.token = token;
        this.timeout = timeout;
    }
}

class QinglongClient {
    constructor(config = new QinglongConfig()) {
        this.config = config;
        this.client = axios.create({
            baseURL: this.config.baseUrl,
            timeout: this.config.timeout,
            headers: {
                'Content-Type': 'application/json',
                'User-Agent': 'Qinglong-NodeJS-Client/1.0.0'
            }
        });
        
        this._setupInterceptor();
    }
    
    _setupInterceptor() {
        // 请求拦截器 - 添加认证token
        this.client.interceptors.request.use((config) => {
            if (this.config.token) {
                config.headers.Authorization = `Bearer ${this.config.token}`;
            }
            return config;
        });
        
        // 响应拦截器 - 统一错误处理
        this.client.interceptors.response.use(
            (response) => {
                const { data } = response;
                if (data.code !== 200) {
                    throw new QinglongException(data.message || 'API调用失败', data.code);
                }
                return data;
            },
            (error) => {
                if (error.response) {
                    throw new QinglongException(
                        `HTTP ${error.response.status}: ${error.response.statusText}`,
                        error.response.status
                    );
                } else if (error.request) {
                    throw new QinglongException('网络请求失败，请检查服务器连接');
                } else {
                    throw new QinglongException(error.message);
                }
            }
        );
    }
    
    async login(username, password) {
        try {
            const response = await this.client.post('/user/login', {
                username,
                password
            });
            
            this.config.token = response.data.token;
            // 更新后续请求的token
            this.client.defaults.headers.Authorization = `Bearer ${this.config.token}`;
            return this.config.token;
        } catch (error) {
            throw new QinglongException(`登录失败: ${error.message}`);
        }
    }
    
    // 定时任务管理
    async listCrons(searchValue = null) {
        const params = {};
        if (searchValue) params.searchValue = searchValue;
        
        const response = await this.client.get('/crons', { params });
        return response.data;
    }
    
    async createCron(taskData) {
        const requiredFields = ['name', 'command', 'schedule'];
        for (const field of requiredFields) {
            if (!taskData[field]) {
                throw new QinglongException(`缺少必要字段: ${field}`);
            }
        }
        
        const response = await this.client.post('/crons', taskData);
        return response.data;
    }
    
    async runCron(taskIds) {
        const response = await this.client.put('/crons/run', taskIds);
        return response.data;
    }
    
    async stopCron(taskIds) {
        const response = await this.client.put('/crons/stop', taskIds);
        return response.data;
    }
    
    async deleteCron(taskIds) {
        const response = await this.client.delete('/crons', { data: taskIds });
        return response.data;
    }
    
    // 环境变量管理
    async listEnvs(searchValue = null) {
        const params = {};
        if (searchValue) params.searchValue = searchValue;
        
        const response = await this.client.get('/envs', { params });
        return response.data;
    }
    
    async createEnv(envData) {
        const response = await this.client.post('/envs', [envData]);
        return response.data[0];
    }
    
    async updateEnv(envId, envData) {
        const data = { ...envData, id: envId };
        const response = await this.client.put('/envs', data);
        return response.data;
    }
    
    async deleteEnv(envIds) {
        const response = await this.client.delete('/envs', { data: envIds });
        return response.data;
    }
    
    // 脚本管理
    async listScripts(scriptPath = null) {
        const params = {};
        if (scriptPath) params.path = scriptPath;
        
        const response = await this.client.get('/scripts', { params });
        return response.data;
    }
    
    async getScriptContent(filePath, fileName) {
        const params = {
            path: filePath,
            file: fileName
        };
        const response = await this.client.get('/scripts/detail', { params });
        return response.data;
    }
    
    async runScript(filePath, fileName, content = '') {
        const data = {
            filename: fileName,
            path: filePath,
            content: content
        };
        const response = await this.client.put('/scripts/run', data);
        return response.data;
    }
    
    // 文件上传功能
    async uploadScript(filePath, fileName, content) {
        const formData = new FormData();
        const blob = new Blob([content], { type: 'text/plain' });
        formData.append('file', blob, fileName);
        formData.append('filename', fileName);
        formData.append('path', filePath);
        
        const response = await this.client.post('/scripts', formData, {
            headers: {
                'Content-Type': 'multipart/form-data'
            }
        });
        return response.data;
    }
}

class QinglongException extends Error {
    constructor(message, code = 500) {
        super(message);
        this.name = 'QinglongException';
        this.code = code;
    }
}

module.exports = { QinglongClient, QinglongConfig, QinglongException };

使用示例

// 配置和使用示例
const { QinglongClient, QinglongConfig } = require('./qinglong-client');
require('dotenv').config();

async function main() {
    // 配置客户端
    const config = new QinglongConfig(
        process.env.QL_URL || 'http://localhost:5700',
        process.env.QL_TOKEN
    );
    
    const client = new QinglongClient(config);
    
    // 登录（如果没有token）
    if (!config.token) {
        try {
            const token = await client.login(
                process.env.QL_USERNAME || 'admin',
                process.env.QL_PASSWORD || 'admin'
            );
            console.log('登录成功，获取到token:', token);
        } catch (error) {
            console.error('登录失败:', error.message);
            return;
        }
    }
    
    // 定时任务管理
    async function manageCronTasks() {
        try {
            // 获取所有任务
            const tasks = await client.listCrons();
            console.log(`当前有 ${tasks.length} 个定时任务`);
            
            // 创建新任务
            const newTask = await client.createCron({
                name: 'Node.js数据同步',
                command: 'node /scripts/sync-data.js',
                schedule: '0 3 * * *',
                labels: ['sync', 'nodejs']
            });
            console.log('创建任务成功:', newTask.id);
            
            // 立即运行任务
            await client.runCron([newTask.id]);
            console.log('任务已开始运行');
            
        } catch (error) {
            console.error('定时任务管理失败:', error.message);
        }
    }
    
    // 环境变量管理
    async function manageEnvironmentVariables() {
        try {
            const envs = await client.listEnvs();
            console.log(`当前有 ${envs.length} 个环境变量`);
            
            // 创建API密钥环境变量
            const apiKeyEnv = await client.createEnv({
                name: 'API_SECRET',
                value: 'your-secret-key-here',
                remarks: '第三方API认证密钥'
            });
            console.log('环境变量创建成功:', apiKeyEnv.id);
            
        } catch (error) {
            console.error('环境变量管理失败:', error.message);
        }
    }
    
    // 脚本操作
    async function manageScripts() {
        try {
            const scripts = await client.listScripts();
            console.log('脚本列表:', scripts);
            
            // 读取脚本内容
            const content = await client.getScriptContent('', 'example.js');
            console.log('脚本内容长度:', content.length);
            
            // 运行脚本
            const result = await client.runScript('', 'example.js');
            console.log('脚本执行结果:', result);
            
        } catch (error) {
            console.error('脚本管理失败:', error.message);
        }
    }
    
    // 执行所有操作
    await manageCronTasks();
    await manageEnvironmentVariables();
    await manageScripts();
}

main().catch(console.error);

高级用法和最佳实践

1. 错误处理和重试机制

# Python重试装饰器
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((QinglongException, requests.exceptions.RequestException))
)
def safe_api_call(self, method, endpoint, **kwargs):
    """带重试机制的API调用"""
    return self._request(method, endpoint, **kwargs)

2. 批量操作支持

// Node.js批量操作示例
async function batchCreateTasks(tasks) {
    const results = [];
    for (const task of tasks) {
        try {
            const result = await client.createCron(task);
            results.push({ success: true, data: result });
        } catch (error) {
            results.push({ success: false, error: error.message });
        }
        // 添加延迟避免请求过于频繁
        await new Promise(resolve => setTimeout(resolve, 100));
    }
    return results;
}

3. 实时监控和WebSocket集成

# WebSocket实时日志监控
import websocket
import json
import threading

class QinglongWebSocketClient:
    def __init__(self, base_url, token):
        self.ws_url = f"ws://{base_url.replace('http://', '').replace('https://', '')}/ws"
        self.token = token
        self.ws = None
        
    def on_message(self, ws, message):
        """处理WebSocket消息"""
        data = json.loads(message)
        if data.get('type') == 'log':
            print(f"[LOG] {data.get('data', '')}")
        elif data.get('type') == 'cron':
            print(f"[CRON] 任务状态更新: {data.get('data', {})}")
    
    def connect(self):
        """连接WebSocket"""
        self.ws = websocket.WebSocketApp(
            self.ws_url,
            header={'Authorization': f'Bearer {self.token}'},
            on_message=self.on_message
        )
        
        # 在后台线程中运行
        wst = threading.Thread(target=self.ws.run_forever)
        wst.daemon = True
        wst.start()

4. 配置管理和环境隔离

# config.yaml - 多环境配置
environments:
  development:
    base_url: "http://localhost:5700"
    token: "dev-token-123"
    script_path: "/scripts/dev"
  
  production:
    base_url: "https://qinglong.prod.com"
    token: "prod-token-456"
    script_path: "/scripts/prod"
  
  staging:
    base_url: "https://qinglong.staging.com"
    token: "staging-token-789"
    script_path: "/scripts/staging"

完整项目结构

qinglong-client/
├── python/
│   ├── qinglong_client.py      # 主客户端类
│   ├── exceptions.py           # 异常类
│   ├── models.py               # 数据模型
│   ├── utils.py                # 工具函数
│   └── examples/
│       ├── basic_usage.py      # 基础使用示例
│       ├── advanced_usage.py   # 高级用法示例
│       └── batch_operations.py # 批量操作示例
├── nodejs/
│   ├── package.json
│   ├── qinglong-client.js      # 主客户端类
│   ├── exceptions.js           # 异常类
│   └── examples/
│       ├── basic-usage.js      # 基础使用示例
│       ├── advanced-usage.js   # 高级用法示例
│       └── batch-operations.js # 批量操作示例
├── config/
│   ├── development.yaml        # 开发环境配置
│   ├── production.yaml         # 生产环境配置
│   └── .env.example            # 环境变量示例
└── README.md                   # 项目文档

性能优化建议

1. 连接池管理: 复用HTTP连接减少TCP握手开销
2. 请求批处理: 合并多个小请求为批量操作
3. 缓存策略: 对不经常变化的数据实施缓存
4. 异步处理: 使用异步IO提高并发性能
5. 监控告警: 集成Prometheus等监控系统

安全注意事项

1. Token保护: 不要将token硬编码在代码中，使用环境变量或配置管理
2. 权限控制: 遵循最小权限原则，使用只读token where possible
3. 输入验证: 对所有用户输入进行严格的验证和清理
4. HTTPS加密: 在生产环境始终使用HTTPS协议
5. 日志脱敏: 避免在日志中记录敏感信息

总结

通过本文介绍的Python和Node.js客户端，你可以轻松实现：

• 🚀 自动化任务管理: 批量创建、运行、监控定时任务
• 🔧 环境配置管理: 动态管理环境变量和配置
• 📁 脚本操作: 上传、下载、执行脚本文件
• 📊 状态监控: 实时获取系统状态和任务执行情况
• 🔒 安全集成: 安全的认证和权限管理

这些客户端为将qinglong集成到你的自动化工作流提供了强大的工具基础，无论是CI/CD流水线、数据管道还是运维自动化场景都能得心应手。

下一步行动:

1. 根据你的技术栈选择合适的客户端版本
2. 配置环境变量和认证信息
3. 从简单的定时任务管理开始尝试
4. 逐步扩展到复杂的自动化场景

开始你的qinglong自动化之旅吧！

【免费下载链接】qinglong 支持 Python3、JavaScript、Shell、Typescript 的定时任务管理平台（Timed task management platform supporting Python3, JavaScript, Shell, Typescript） 项目地址: https://gitcode.com/GitHub_Trending/qi/qinglong