Upload 4 files

Dockerfile

@@ -0,0 +1,34 @@
+# Use Python 3.10 slim image as base
+FROM python:3.10-slim
+
+# Create non-root user
+RUN useradd -m -u 1000 user
+
+# Set environment variables
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH \
+    PORT=7860
+
+# Create and set working directory
+WORKDIR $HOME/app
+
+# Copy requirements first to leverage Docker cache
+COPY --chown=user requirements.txt .
+
+# Switch to non-root user
+USER user
+
+# Install dependencies
+RUN pip install --user -r requirements.txt
+
+# Copy application files
+COPY --chown=user . .
+
+# Create uploads directory
+RUN mkdir -p uploads
+
+# Expose port
+EXPOSE ${PORT}
+
+# Command to run the application
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860", "--reload"]

client.py

@@ -0,0 +1,121 @@
+import requests
+import os
+
+class FileServerClient:
+    def __init__(self, base_url="http://localhost:7860"):
+        self.base_url = base_url
+        self.access_codes = {}  # Store filename to access code mapping
+
+    def upload_file(self, file_path):
+        """Upload a single file to the server"""
+        if not os.path.exists(file_path):
+            print(f"Error: File {file_path} not found")
+            return
+        
+        with open(file_path, 'rb') as f:
+            files = {'file': f}
+            response = requests.post(f"{self.base_url}/upload/", files=files)
+            result = response.json()
+            
+            # Store the access code
+            if 'access_code' in result:
+                self.access_codes[os.path.basename(file_path)] = result['access_code']
+                
+        return result
+
+    def upload_multiple_files(self, file_paths):
+        """Upload multiple files to the server"""
+        files = []
+        for file_path in file_paths:
+            if os.path.exists(file_path):
+                files.append(('files', open(file_path, 'rb')))
+            else:
+                print(f"Warning: File {file_path} not found, skipping...")
+        
+        response = requests.post(f"{self.base_url}/upload-multiple/", files=files)
+        result = response.json()
+        
+        # Store access codes
+        if 'files' in result:
+            for file_info in result['files']:
+                self.access_codes[file_info['filename']] = file_info['access_code']
+        
+        # Close all opened files
+        for _, file_obj in files:
+            file_obj.close()
+            
+        return result
+
+    def list_files(self):
+        """List all files on the server"""
+        response = requests.get(f"{self.base_url}/files/")
+        return response.json()
+
+    def download_file(self, access_code, save_path):
+        """Download a file using its access code"""
+        response = requests.get(f"{self.base_url}/download/{access_code}", stream=True)
+        
+        if response.status_code == 200:
+            with open(save_path, 'wb') as f:
+                for chunk in response.iter_content(chunk_size=8192):
+                    f.write(chunk)
+            return True
+        else:
+            print(f"Error: {response.json()['detail']}")
+            return False
+
+    def delete_file(self, access_code):
+        """Delete a file using its access code"""
+        response = requests.delete(f"{self.base_url}/delete/{access_code}")
+        return response.json()
+
+
+def main():
+    # Create client instance
+    client = FileServerClient()
+    
+    print("\n=== File Server Client Demo ===\n")
+
+    # 1. Upload a single file
+    print("1. Testing single file upload:")
+    test_file = "test_upload.txt"
+    with open(test_file, "w") as f:
+        f.write("This is a test file for upload!")
+    
+    result = client.upload_file(test_file)
+    print(result)
+    access_code = result['access_code']
+
+    # 2. List files
+    print("\n2. Listing all files on server:")
+    files = client.list_files()
+    print(files)
+
+    # 3. Download the file using access code
+    print("\n3. Downloading the file:")
+    download_path = "downloaded_test.txt"
+    success = client.download_file(access_code, download_path)
+    if success:
+        print(f"File downloaded successfully to {download_path}")
+
+    # 4. Upload multiple files
+    print("\n4. Testing multiple file upload:")
+    test_file2 = "test_upload2.txt"
+    with open(test_file2, "w") as f:
+        f.write("This is another test file!")
+    
+    result = client.upload_multiple_files([test_file, test_file2])
+    print(result)
+
+    # 5. Delete files using access codes
+    print("\n5. Deleting test files from server:")
+    for file_info in result['files']:
+        print(client.delete_file(file_info['access_code']))
+
+    # Clean up local test files
+    os.remove(test_file)
+    os.remove(test_file2)
+    os.remove(download_path)
+
+if __name__ == "__main__":
+    main() 

main.py

@@ -0,0 +1,119 @@
+from fastapi import FastAPI, UploadFile, File, HTTPException
+from fastapi.responses import FileResponse
+from typing import List, Dict
+import os
+import shutil
+import uuid
+
+app = FastAPI()
+
+# Create uploads directory if it doesn't exist
+UPLOAD_DIR = "uploads"
+if not os.path.exists(UPLOAD_DIR):
+    os.makedirs(UPLOAD_DIR)
+
+# Store file mappings: {unique_code: filename}
+file_codes: Dict[str, str] = {}
+# Reverse mapping for easy lookup: {filename: unique_code}
+filename_codes: Dict[str, str] = {}
+
+@app.post("/upload/")
+async def upload_file(file: UploadFile = File(...)):
+    try:
+        # Generate unique code
+        unique_code = str(uuid.uuid4())[:8]
+        
+        # Save the uploaded file
+        file_path = os.path.join(UPLOAD_DIR, file.filename)
+        with open(file_path, "wb") as buffer:
+            shutil.copyfileobj(file.file, buffer)
+        
+        # Store the mapping
+        file_codes[unique_code] = file.filename
+        filename_codes[file.filename] = unique_code
+        
+        return {
+            "filename": file.filename,
+            "access_code": unique_code,
+            "message": "File uploaded successfully"
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/upload-multiple/")
+async def upload_multiple_files(files: List[UploadFile] = File(...)):
+    try:
+        uploaded_files = []
+        for file in files:
+            # Generate unique code for each file
+            unique_code = str(uuid.uuid4())[:8]
+            
+            file_path = os.path.join(UPLOAD_DIR, file.filename)
+            with open(file_path, "wb") as buffer:
+                shutil.copyfileobj(file.file, buffer)
+            
+            # Store the mapping
+            file_codes[unique_code] = file.filename
+            filename_codes[file.filename] = unique_code
+            
+            uploaded_files.append({
+                "filename": file.filename,
+                "access_code": unique_code
+            })
+        return {"files": uploaded_files, "message": "Files uploaded successfully"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/files/")
+async def list_files():
+    try:
+        files = []
+        for filename in os.listdir(UPLOAD_DIR):
+            code = filename_codes.get(filename, "no_code")
+            files.append({
+                "filename": filename,
+                "access_code": code
+            })
+        return {"files": files}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/download/{access_code}")
+async def download_file(access_code: str):
+    try:
+        if access_code not in file_codes:
+            raise HTTPException(status_code=404, detail="Invalid access code")
+        
+        filename = file_codes[access_code]
+        file_path = os.path.join(UPLOAD_DIR, filename)
+        
+        if os.path.exists(file_path):
+            return FileResponse(file_path, filename=filename)
+        else:
+            raise HTTPException(status_code=404, detail="File not found")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.delete("/delete/{access_code}")
+async def delete_file(access_code: str):
+    try:
+        if access_code not in file_codes:
+            raise HTTPException(status_code=404, detail="Invalid access code")
+        
+        filename = file_codes[access_code]
+        file_path = os.path.join(UPLOAD_DIR, filename)
+        
+        if os.path.exists(file_path):
+            os.remove(file_path)
+            # Remove from both mappings
+            del filename_codes[filename]
+            del file_codes[access_code]
+            return {"message": f"File {filename} deleted successfully"}
+        else:
+            raise HTTPException(status_code=404, detail="File not found")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860) 

requirements.txt

@@ -0,0 +1,4 @@
+fastapi==0.104.1
+python-multipart==0.0.6
+uvicorn==0.24.0
+requests==2.31.0