Files
pic/api/routes/files.py
roof 09138fbc18 A5: Extract all route groups into Flask blueprints (app.py -1735 lines)
Extract 9 route groups out of app.py into routes/ blueprints:
- routes/network.py  — DNS, DHCP, NTP, network info/test (10 routes)
- routes/wireguard.py — WireGuard keys, peers, config, enforcement (18 routes)
- routes/cells.py    — cell-to-cell connections (5 routes)
- routes/peers.py    — peer CRUD + IP update + _next_peer_ip helper (10 routes)
- routes/routing.py  — NAT, peer routes, firewall, iptables (17 routes)
- routes/vault.py    — certs, trust, secrets (19 routes)
- routes/containers.py — containers, images, volumes (14 routes)
- routes/services.py — service bus, logs, services status/connectivity (18 routes)
- routes/peer_dashboard.py — peer-scoped dashboard/services (2 routes)

All blueprints use lazy `from app import X` inside route bodies to preserve
test patch compatibility (patch('app.email_manager', mock) still works).

Also included in this commit:
- A1 fix: backup/restore now includes email/calendar user files
- A2 fix: apply_config sets applying=True flag via helper container
- A3 fix: add_peer rolls back firewall on DNS failure

app.py reduced: 3011 → 1294 lines. 1021 tests passing.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-01 06:11:21 -04:00

160 lines
5.7 KiB
Python

import logging
from flask import Blueprint, request, jsonify
logger = logging.getLogger('picell')
bp = Blueprint('files', __name__)
@bp.route('/api/files/users', methods=['GET'])
def get_file_users():
"""Get file storage users."""
try:
from app import file_manager
users = file_manager.get_users()
return jsonify(users)
except Exception as e:
logger.error(f"Error getting file users: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/files/users', methods=['POST'])
def create_file_user():
"""Create file storage user."""
try:
from app import file_manager
data = request.get_json(silent=True)
if data is None:
return jsonify({"error": "No data provided"}), 400
username = data.get('username')
password = data.get('password')
if not username or not password:
return jsonify({"error": "Missing required fields: username, password"}), 400
result = file_manager.create_user(username, password)
return jsonify({"created": result})
except Exception as e:
logger.error(f"Error creating file user: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/files/users/<username>', methods=['DELETE'])
def delete_file_user(username):
"""Delete file storage user."""
try:
from app import file_manager
result = file_manager.delete_user(username)
return jsonify(result)
except Exception as e:
logger.error(f"Error deleting file user: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/files/folders', methods=['POST'])
def create_folder():
"""Create folder."""
try:
from app import file_manager
data = request.get_json(silent=True)
if data is None:
return jsonify({"error": "No data provided"}), 400
username = data.get('username')
folder_path = data.get('folder_path') or data.get('path')
if not username or not folder_path:
return jsonify({"error": "Missing required fields: username, folder_path"}), 400
result = file_manager.create_folder(username, folder_path)
return jsonify({"created": result})
except ValueError as e:
return jsonify({"error": str(e)}), 400
except Exception as e:
logger.error(f"Error creating folder: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/files/folders/<username>/<path:folder_path>', methods=['DELETE'])
def delete_folder(username, folder_path):
"""Delete folder."""
try:
from app import file_manager
result = file_manager.delete_folder(username, folder_path)
return jsonify(result)
except ValueError as e:
return jsonify({"error": str(e)}), 400
except Exception as e:
logger.error(f"Error deleting folder: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/files/upload/<username>', methods=['POST'])
def upload_file(username):
"""Upload file."""
try:
from app import file_manager
if 'file' not in request.files:
return jsonify({"error": "No file provided"}), 400
file = request.files['file']
path = request.form.get('path', '') or file.filename or ''
file_data = file.read()
result = file_manager.upload_file(username, path, file_data)
return jsonify({"uploaded": result})
except ValueError as e:
return jsonify({"error": str(e)}), 400
except Exception as e:
logger.error(f"Error uploading file: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/files/download/<username>/<path:file_path>', methods=['GET'])
def download_file(username, file_path):
"""Download file."""
try:
from app import file_manager
result = file_manager.download_file(username, file_path)
return jsonify(result)
except ValueError as e:
return jsonify({"error": str(e)}), 400
except Exception as e:
logger.error(f"Error downloading file: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/files/delete/<username>/<path:file_path>', methods=['DELETE'])
def delete_file(username, file_path):
"""Delete file."""
try:
from app import file_manager
result = file_manager.delete_file(username, file_path)
return jsonify(result)
except ValueError as e:
return jsonify({"error": str(e)}), 400
except Exception as e:
logger.error(f"Error deleting file: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/files/list/<username>', methods=['GET'])
def list_files(username):
"""List files."""
try:
from app import file_manager
folder = request.args.get('folder', '')
result = file_manager.list_files(username, folder)
return jsonify(result)
except ValueError as e:
return jsonify({"error": str(e)}), 400
except Exception as e:
logger.error(f"Error listing files: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/files/status', methods=['GET'])
def get_file_status():
"""Get file service status."""
try:
from app import file_manager
status = file_manager.get_status()
return jsonify(status)
except Exception as e:
logger.error(f"Error getting file status: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/files/connectivity', methods=['GET'])
def test_file_connectivity():
"""Test file service connectivity."""
try:
from app import file_manager
result = file_manager.test_connectivity()
return jsonify(result)
except Exception as e:
logger.error(f"Error testing file connectivity: {e}")
return jsonify({"error": str(e)}), 500