299 lines
9 KiB
Python
299 lines
9 KiB
Python
"""Routes for file upload and management"""
|
|
|
|
from io import BytesIO
|
|
|
|
from flask import Response, request
|
|
from flask_jwt_extended import jwt_required
|
|
from werkzeug.datastructures import FileStorage
|
|
|
|
from app.decorators import (get_current_user_id, load_card_owned,
|
|
load_comment_owned, load_file_accessible_by_uuid)
|
|
from app.schemas import (FileAttachmentWithUrlResponse, FileDeleteResponse,
|
|
MultipleFilesResponse)
|
|
from app.services.storage.attachment_service import AttachmentService
|
|
from app.services.storage.file_service import FileService
|
|
|
|
from . import kanban_bp
|
|
|
|
|
|
@kanban_bp.route("/files/<string:file_uuid>", methods=["GET"])
|
|
@jwt_required()
|
|
@load_file_accessible_by_uuid
|
|
def get_file(file_uuid, file):
|
|
"""Get file information with UUID-based URLs"""
|
|
attachment_service = AttachmentService()
|
|
file_info = attachment_service.get_attachment_with_url(file.id)
|
|
|
|
return file_info, 200
|
|
|
|
|
|
@kanban_bp.route("/files/<string:file_uuid>", methods=["DELETE"])
|
|
@jwt_required()
|
|
@load_file_accessible_by_uuid
|
|
def delete_file(file_uuid, file):
|
|
"""Delete a file"""
|
|
attachment_service = AttachmentService()
|
|
success = attachment_service.delete_attachment(file.id)
|
|
|
|
if not success:
|
|
return {"error": "Failed to delete file"}, 500
|
|
|
|
return FileDeleteResponse(message="File deleted", file_id=file.id).model_dump(), 200
|
|
|
|
|
|
@kanban_bp.route("/cards/<int:card_id>/attachments", methods=["GET"])
|
|
@jwt_required()
|
|
@load_card_owned
|
|
def get_card_attachments(card_id, card):
|
|
"""Get all attachments for a card"""
|
|
attachment_service = AttachmentService()
|
|
attachments = attachment_service.get_card_attachments(card_id)
|
|
|
|
# Get URLs for each attachment
|
|
files_with_urls = []
|
|
for attachment in attachments:
|
|
file_info = attachment_service.get_attachment_with_url(attachment.id)
|
|
files_with_urls.append(file_info)
|
|
|
|
return (
|
|
MultipleFilesResponse(
|
|
files=files_with_urls,
|
|
count=len(attachments),
|
|
).model_dump(),
|
|
200,
|
|
)
|
|
|
|
|
|
@kanban_bp.route("/cards/<int:card_id>/attachments", methods=["POST"])
|
|
@jwt_required()
|
|
@load_card_owned
|
|
def upload_card_attachment(card_id, card):
|
|
"""Upload a file to a specific card"""
|
|
return _upload_file(entity_type="Card", entity_id=card_id)
|
|
|
|
|
|
@kanban_bp.route("/comments/<int:comment_id>/attachments", methods=["GET"])
|
|
@jwt_required()
|
|
@load_comment_owned
|
|
def get_comment_attachments(comment_id, comment):
|
|
"""Get all attachments for a comment"""
|
|
attachment_service = AttachmentService()
|
|
attachments = attachment_service.get_comment_attachments(comment_id)
|
|
|
|
# Get URLs for each attachment
|
|
files_with_urls = []
|
|
for attachment in attachments:
|
|
file_info = attachment_service.get_attachment_with_url(attachment.id)
|
|
files_with_urls.append(file_info)
|
|
|
|
return (
|
|
MultipleFilesResponse(
|
|
files=files_with_urls,
|
|
count=len(attachments),
|
|
).model_dump(),
|
|
200,
|
|
)
|
|
|
|
|
|
@kanban_bp.route("/comments/<int:comment_id>/attachments", methods=["POST"])
|
|
@jwt_required()
|
|
@load_comment_owned
|
|
def upload_comment_attachment(comment_id, comment):
|
|
"""Upload a file to a specific comment"""
|
|
return _upload_file(entity_type="Comment", entity_id=comment_id)
|
|
|
|
|
|
@kanban_bp.route("/files/upload", methods=["POST"])
|
|
@jwt_required()
|
|
def upload_file():
|
|
"""Upload a file to an entity (Card, Comment, etc.)"""
|
|
# Validate request
|
|
entity_type = request.form.get("entity_type")
|
|
entity_id = request.form.get("entity_id")
|
|
|
|
if not entity_type or not entity_id:
|
|
return {"error": "entity_type and entity_id are required"}, 400
|
|
|
|
# Validate entity_type
|
|
valid_types = ["Card", "Comment", "Epic", "Board"]
|
|
if entity_type not in valid_types:
|
|
return {
|
|
"error": f"Invalid entity_type: {entity_type}. Must be one of {valid_types}"
|
|
}, 400
|
|
|
|
return _upload_file(entity_type=entity_type, entity_id=int(entity_id))
|
|
|
|
|
|
def _upload_file(entity_type: str, entity_id: int):
|
|
"""
|
|
Helper function to handle file upload
|
|
|
|
Args:
|
|
entity_type: Type of entity (Card, Comment, etc.)
|
|
entity_id: ID of the entity
|
|
|
|
Returns:
|
|
tuple: (response_dict, status_code)
|
|
"""
|
|
# Get file from request
|
|
if "file" not in request.files:
|
|
return {"error": "No file provided"}, 400
|
|
|
|
file: FileStorage = request.files["file"]
|
|
if file.filename == "":
|
|
return {"error": "No file selected"}, 400
|
|
|
|
# Get user ID
|
|
user_id = get_current_user_id()
|
|
|
|
# Create BytesIO from file
|
|
file_data = BytesIO()
|
|
file.save(file_data)
|
|
file_size = len(file_data.getvalue())
|
|
|
|
# Get MIME type
|
|
mime_type = file.mimetype or "application/octet-stream"
|
|
|
|
# Upload file
|
|
attachment_service = AttachmentService()
|
|
attachment = attachment_service.attach_file_to_entity(
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
file_data=file_data,
|
|
filename=file.filename,
|
|
content_type=mime_type,
|
|
file_size=file_size,
|
|
uploaded_by=user_id,
|
|
)
|
|
|
|
# Get attachment with URLs for response
|
|
file_info = attachment_service.get_attachment_with_url(attachment.id)
|
|
return FileAttachmentWithUrlResponse(**file_info).model_dump(), 201
|
|
|
|
|
|
@kanban_bp.route("/files/<string:file_uuid>/download", methods=["GET"])
|
|
@jwt_required()
|
|
@load_file_accessible_by_uuid
|
|
def download_file(file_uuid, file):
|
|
"""
|
|
Download a file through secure Flask proxy using UUID
|
|
|
|
This route serves as a proxy to MinIO, hiding the MinIO server
|
|
from the frontend and ensuring authentication/authorization.
|
|
|
|
Args:
|
|
file_uuid: UUID of the file attachment
|
|
file: FileAttachment (injected by @load_file_accessible_by_uuid decorator)
|
|
|
|
Returns:
|
|
Response: File data with proper headers
|
|
"""
|
|
file_service = FileService()
|
|
|
|
# Download file from MinIO
|
|
file_data = file_service.storage.download_file(
|
|
file.minio_bucket, file.minio_object_name
|
|
)
|
|
|
|
# Get file info for proper headers
|
|
file_info = file_service.storage.get_file_info(
|
|
file.minio_bucket, file.minio_object_name
|
|
)
|
|
|
|
# Create Flask Response with file data
|
|
return Response(
|
|
file_data.getvalue(),
|
|
mimetype=file.mime_type,
|
|
headers={
|
|
"Content-Disposition": f'attachment; filename="{file.original_name}"',
|
|
"Content-Length": str(file_info["size"]),
|
|
"Cache-Control": "private, max-age=3600", # Cache for 1 hour
|
|
},
|
|
)
|
|
|
|
|
|
@kanban_bp.route("/files/<string:file_uuid>/thumbnail", methods=["GET"])
|
|
@jwt_required()
|
|
@load_file_accessible_by_uuid
|
|
def download_thumbnail(file_uuid, file):
|
|
"""
|
|
Download a file thumbnail through secure Flask proxy using UUID
|
|
|
|
This route serves as a proxy to MinIO for thumbnails,
|
|
ensuring authentication with JWT.
|
|
|
|
Args:
|
|
file_uuid: UUID of the file attachment
|
|
file: FileAttachment (injected by @load_file_accessible_by_uuid decorator)
|
|
|
|
Returns:
|
|
Response: Thumbnail image data with proper headers
|
|
"""
|
|
# Check if thumbnail exists
|
|
if not file.thumbnail_minio_bucket or not file.thumbnail_minio_object_name:
|
|
return {"error": "Thumbnail not available"}, 404
|
|
|
|
file_service = FileService()
|
|
|
|
# Download thumbnail from MinIO
|
|
thumbnail_data = file_service.storage.download_file(
|
|
file.thumbnail_minio_bucket,
|
|
file.thumbnail_minio_object_name,
|
|
)
|
|
|
|
# Get thumbnail info for proper headers
|
|
thumb_info = file_service.storage.get_file_info(
|
|
file.thumbnail_minio_bucket,
|
|
file.thumbnail_minio_object_name,
|
|
)
|
|
|
|
# Create Flask Response with thumbnail data
|
|
return Response(
|
|
thumbnail_data.getvalue(),
|
|
mimetype="image/jpeg",
|
|
headers={
|
|
"Content-Length": str(thumb_info["size"]),
|
|
"Cache-Control": "private, max-age=3600",
|
|
},
|
|
)
|
|
|
|
|
|
@kanban_bp.route("/files/<string:file_uuid>/view", methods=["GET"])
|
|
@jwt_required()
|
|
@load_file_accessible_by_uuid
|
|
def view_file(file_uuid, file):
|
|
"""
|
|
View a file inline through secure Flask proxy using UUID
|
|
|
|
Similar to download, but uses inline disposition so browsers
|
|
display the file instead of downloading it.
|
|
|
|
Args:
|
|
file_uuid: UUID of the file attachment
|
|
file: FileAttachment (injected by @load_file_accessible_by_uuid decorator)
|
|
|
|
Returns:
|
|
Response: File data with proper headers
|
|
"""
|
|
file_service = FileService()
|
|
|
|
# Download file from MinIO
|
|
file_data = file_service.storage.download_file(
|
|
file.minio_bucket, file.minio_object_name
|
|
)
|
|
|
|
# Get file info for proper headers
|
|
file_info = file_service.storage.get_file_info(
|
|
file.minio_bucket, file.minio_object_name
|
|
)
|
|
|
|
# Create Flask Response with file data (inline disposition)
|
|
return Response(
|
|
file_data.getvalue(),
|
|
mimetype=file.mime_type,
|
|
headers={
|
|
"Content-Disposition": f'inline; filename="{file.original_name}"',
|
|
"Content-Length": str(file_info["size"]),
|
|
"Cache-Control": "private, max-age=3600", # Cache for 1 hour
|
|
},
|
|
)
|