file-upload [LACORTE]

.env

@@ -0,0 +1,12 @@
+# Flask Environment
+FLASK_ENV=development
+UPLOAD_FOLDER=uploads
+
+# Database Configuration
+DB_USER=root
+DB_PASSWORD=hasler619
+DB_HOST=localhost
+DB_PORT=3306
+DB_NAME=flask_uploads_db
+SQLALCHEMY_DATABASE_URI=mysql+pymysql://root:hasler619@localhost:3306/flask_uploads_db
+# SQLALCHEMY_TRACK_MODIFICATIONS=False

.vscode/settings.json

@@ -0,0 +1,5 @@
+{
+    "python.analysis.extraPaths": [
+        "./models"
+    ]
+}

app.py

@@ -0,0 +1,33 @@
+from flask import Flask
+from config import Config
+from extensions import db
+from controllers.file_controller import file_bp
+from models.uploaded_file import UploadedFile # 
+import logging, os
+
+# forda logs
+log_folder = "logs"
+os.makedirs(log_folder, exist_ok=True)
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    handlers=[
+        logging.FileHandler(os.path.join(log_folder, "app.log")),
+        logging.StreamHandler()
+    ]
+)
+
+def create_app():
+    app = Flask(__name__)
+    app.config.from_object(Config)
+    db.init_app(app)
+    app.register_blueprint(file_bp)
+    with app.app_context():
+        db.create_all()
+    return app
+
+app = create_app()
+
+if __name__ == '__main__':
+    app.run(debug=True)

config.py

@@ -0,0 +1,13 @@
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+class Config:
+    SQLALCHEMY_DATABASE_URI = (
+        f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@"
+        f"{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"
+    )
+    SQLALCHEMY_TRACK_MODIFICATIONS = False
+    UPLOAD_FOLDER = os.getenv('UPLOAD_FOLDER', 'uploads')
+    DEBUG = True

controllers/file_controller.py

@@ -0,0 +1,65 @@
+from flask import Blueprint, request, jsonify, send_file, current_app
+from services.file_service import save_file, update_file, delete_file
+from models.uploaded_file import UploadedFile
+import uuid, os, logging
+
+file_bp = Blueprint('file_bp', __name__)
+
+# POST — Upload
+@file_bp.route('/upload', methods=['POST'])
+def upload_file():
+    try:
+        if 'file' not in request.files:
+            return jsonify({'message': 'No file uploaded'}), 400
+        file = request.files['file']
+        folder = os.path.join(current_app.config['UPLOAD_FOLDER'], uuid.uuid4().hex)
+        uploaded = save_file(file, folder)
+        return jsonify({'message': 'Upload successful','id': uploaded.id,'path': uploaded.filepath}), 201
+    except Exception as e:
+        logging.error(f"Upload error: {e}")
+        return jsonify({'message': str(e)}), 400
+
+# GET — Download
+@file_bp.route('/file/<int:file_id>', methods=['GET'])
+def get_file(file_id):
+    try:
+        file_record = UploadedFile.query.get(file_id)
+        if not file_record:
+            return jsonify({'message': 'File not found'}), 404
+        return send_file(file_record.filepath, as_attachment=True)
+    except Exception as e:
+        logging.error(f"Get error: {e}")
+        return jsonify({'message': str(e)}), 400
+
+# PUT — Update
+@file_bp.route('/file/<int:file_id>', methods=['PUT'])
+def update_uploaded_file(file_id):
+    try:
+        if 'file' not in request.files:
+            return jsonify({"message": "No file part"}), 400
+
+        file = request.files['file']
+        if file.filename == '':
+            return jsonify({"message": "No selected file"}), 400
+
+        updated_record = update_file(file_id, file)
+        return jsonify({
+            "message": "File updated successfully",
+            "id": updated_record.id,
+            "path": updated_record.filepath
+        }), 200
+
+    except Exception as e:
+        logging.error(str(e))
+        return jsonify({"message": str(e)}), 500
+
+# DELETE — Delete
+@file_bp.route('/file/<int:file_id>', methods=['DELETE'])
+def delete_uploaded_file(file_id):
+    try:
+        delete_file(file_id)
+        return jsonify({"message": "File deleted successfully"}), 200
+    except Exception as e:
+        logging.error(f"Delete error: {e}")
+        return jsonify({"message": str(e)}), 400
+

extensions.py

@@ -0,0 +1,2 @@
+from flask_sqlalchemy import SQLAlchemy
+db = SQLAlchemy()

models/uploaded_file.py

@@ -0,0 +1,6 @@
+from extensions import db
+
+class UploadedFile(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    filename = db.Column(db.String(255), nullable=False)
+    filepath = db.Column(db.String(512), nullable=False)

services/file_service.py

@@ -0,0 +1,82 @@
+import os
+import uuid
+import logging
+from werkzeug.utils import secure_filename
+from extensions import db
+from models.uploaded_file import UploadedFile
+
+ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'pdf', 'txt'}
+
+def allowed_file(filename):
+    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
+
+def save_file(file, base_folder):
+    if not allowed_file(file.filename):
+        raise ValueError("File type not allowed")
+
+    os.makedirs(base_folder, exist_ok=True)
+    filename = secure_filename(file.filename)
+    unique_name = f"{uuid.uuid4().hex}_{filename}"
+    save_path = os.path.join(base_folder, unique_name)
+    file.save(save_path)
+
+    new_file = UploadedFile(filename=filename, filepath=save_path)
+    db.session.add(new_file)
+    db.session.commit()
+
+    logging.info(f"File saved: {save_path}")
+    return new_file
+
+def update_file(file_id, file):
+    record = UploadedFile.query.get(file_id)
+    if not record:
+        raise ValueError(f"File with ID {file_id} not found")
+
+    if not allowed_file(file.filename):
+        raise ValueError("File type not allowed")
+
+    # delete old file if it exists
+    if os.path.exists(record.filepath):
+        os.remove(record.filepath)
+        logging.info(f"Old file removed: {record.filepath}")
+
+    # save the new file in the same folder
+    folder = os.path.dirname(record.filepath)
+    filename = secure_filename(file.filename)
+    save_path = os.path.join(folder, filename)
+    file.save(save_path)
+
+    # update database record
+    record.filename = filename
+    record.filepath = save_path
+    db.session.commit()
+
+    logging.info(f"File updated: ID={file_id}, Path={save_path}")
+    return record
+
+from extensions import db
+from models.uploaded_file import UploadedFile
+import os, logging
+
+def delete_file(file_id):
+    record = UploadedFile.query.get(file_id)
+    if not record:
+        raise ValueError(f"File with ID {file_id} not found")
+
+    if record.filepath and os.path.exists(record.filepath):
+        os.remove(record.filepath)
+        logging.info(f"File deleted from storage: {record.filepath}")
+
+        folder = os.path.dirname(record.filepath)
+        if os.path.exists(folder) and not os.listdir(folder):
+            os.rmdir(folder)
+            logging.info(f"Empty folder removed: {folder}")
+
+    else:
+        logging.warning(f"File path not found or already deleted: {record.filepath}")
+
+    db.session.delete(record)
+    db.session.commit()
+    logging.info(f"Database record deleted: ID={file_id}")
+
+    return True