(username: str)
| 32 | |
| 33 | @file_router.get('/download/{username}') |
| 34 | async def download_workspace(username: str): |
| 35 | try: |
| 36 | user_workspace = get_or_create_workspace(username) |
| 37 | |
| 38 | # Create a zip file from the user's workspace directory |
| 39 | zip_filename = f'{username}_workspace.zip' |
| 40 | zip_buffer = io.BytesIO() # in-memory buffer to hold the zip file |
| 41 | |
| 42 | # Create a ZipFile object in write mode |
| 43 | with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file: |
| 44 | # Walk through the workspace directory and add files to the zip |
| 45 | for root, dirs, files in os.walk(user_workspace): |
| 46 | for file in files: |
| 47 | file_path = os.path.join(root, file) |
| 48 | arcname = os.path.relpath(file_path, user_workspace) # Store files relative to the workspace |
| 49 | zip_file.write(file_path, arcname) |
| 50 | |
| 51 | # Seek to the beginning of the buffer before sending it |
| 52 | zip_buffer.seek(0) |
| 53 | |
| 54 | # Return the zip file as a Response, without triggering stat checks |
| 55 | return Response( |
| 56 | zip_buffer.read(), # Read the content of the BytesIO object |
| 57 | media_type="application/zip", # Set the media type to zip file |
| 58 | headers={"Content-Disposition": f"attachment; filename={zip_filename}"} |
| 59 | ) |
| 60 | |
| 61 | except Exception as e: |
| 62 | print(f"Error creating zip: {e}") |
| 63 | raise HTTPException(status_code=500, detail=f"Failed to create zip file: {str(e)}") |
| 64 | |
| 65 | # Route to handle file uploads with username |
| 66 | @file_router.post('/upload/{username}') |
nothing calls this directly
no test coverage detected