78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
import functools
|
|
from flask import Blueprint, request, jsonify
|
|
from app.models import ApiKey
|
|
from app.utils.octoprint_client import OctoPrintClient
|
|
from app.models import SystemConfig
|
|
|
|
api_bp = Blueprint('api_handle', __name__, url_prefix='/api/v1')
|
|
|
|
def get_octo_client():
|
|
url = SystemConfig.query.filter_by(key='octoprint_url').first()
|
|
apikey = SystemConfig.query.filter_by(key='octoprint_apikey').first()
|
|
if url and url.value and apikey and apikey.value:
|
|
return OctoPrintClient(url.value, apikey.value)
|
|
return None
|
|
|
|
def require_api_key(f):
|
|
@functools.wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
api_key_header = request.headers.get('X-Api-Key')
|
|
if not api_key_header:
|
|
return jsonify({'error': 'Missing API Key in headers (X-Api-Key)'}), 401
|
|
|
|
key_record = ApiKey.query.filter_by(key=api_key_header).first()
|
|
if not key_record:
|
|
return jsonify({'error': 'Invalid API Key'}), 401
|
|
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
@api_bp.route('/status', methods=['GET'])
|
|
@require_api_key
|
|
def get_status():
|
|
client = get_octo_client()
|
|
if not client:
|
|
return jsonify({'error': 'Printer not configured'}), 503
|
|
try:
|
|
status_data = client.get_printer_status()
|
|
job_data = client.get_job_info()
|
|
return jsonify({'status': status_data, 'job': job_data})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@api_bp.route('/octoprint_client', methods=['POST'])
|
|
@require_api_key
|
|
def invoke_octoprint_client():
|
|
"""
|
|
Expects JSON payload like:
|
|
{
|
|
"method": "pause_print",
|
|
"kwargs": {"action": "pause"}
|
|
}
|
|
"""
|
|
client = get_octo_client()
|
|
if not client:
|
|
return jsonify({'error': 'Printer not configured'}), 503
|
|
|
|
data = request.get_json()
|
|
if not data or 'method' not in data:
|
|
return jsonify({'error': 'Missing method in JSON payload'}), 400
|
|
|
|
method_name = data['method']
|
|
kwargs = data.get('kwargs', {})
|
|
args = data.get('args', [])
|
|
|
|
if not hasattr(client, method_name):
|
|
return jsonify({'error': f'Method {method_name} not found on OctoPrintClient'}), 400
|
|
|
|
func = getattr(client, method_name)
|
|
if not callable(func) or method_name.startswith('_'):
|
|
return jsonify({'error': f'Method {method_name} is not allowed'}), 403
|
|
|
|
try:
|
|
result = func(*args, **kwargs)
|
|
return jsonify({'success': True, 'result': result})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|