### Run Backend Server Example Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Instructions to run the backend server for the TusPyServer example using 'uv'. Ensure you are in the 'example/backend' directory. ```bash uv run server.py ``` -------------------------------- ### Install tuspyserver from source Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Clone the repository and install tuspyserver directly from the source code. ```bash git clone https://github.com/edihasaj/tuspyserver cd tuspyserver pip install . ``` -------------------------------- ### Install tuspyserver with pip Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Install the latest release of tuspyserver using pip. ```bash pip install tuspyserver ``` -------------------------------- ### Run Frontend Development Server Example Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Instructions to run the frontend development server for the TusPyServer example using 'npm'. Ensure you are in the 'example/frontend' directory. ```bash npm run dev ``` -------------------------------- ### Basic FastAPI app setup with tuspyserver Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Integrate the tuspyserver router into a FastAPI application, including CORS middleware and static file serving. ```python from tuspyserver import create_tus_router from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles import uvicorn # initialize a FastAPI app app = FastAPI() # configure cross-origin middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], expose_headers=[ "Location", "Upload-Offset", "Tus-Resumable", "Tus-Version", "Tus-Extension", "Tus-Max-Size", "Upload-Expires", "Upload-Length", ], ) # use completion hook to log uploads def log_upload(file_path: str, metadata: dict): print("Upload complete") print(file_path) print(metadata) # mount the tus router to our app.include_router( create_tus_router( files_dir="./uploads", on_upload_complete=log_upload, ) ) ``` -------------------------------- ### File Routing Dependency Injection Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Utilize dependency injection with a file dependency to directly store files. This example shows how to retrieve file directory and UID based on user and metadata. ```python from fastapi import Depends, HTTPException from your_app.dependencies import get_db, get_current_user, get_user_dir def get_file( db=Depends(get_db), current_user=Depends(get_current_user), ) -> Callable[[dict, dict], None]: # callback function async def handler(metadata: dict): # Get the file name file_name = metadata["file_name"] # Get the file directory file_dir = get_user_dir(current_user) return { "file_dir": file_dir, "uid": file_name } return handler # Include router with the pre-create DI hook app.include_router( create_tus_router( file_dep=file_dep, ) ) ``` -------------------------------- ### Create tus router with default options Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Initialize the tus router with basic configuration, specifying the storage directory and a completion callback. ```python from tuspyserver import create_tus_router tus_router = create_tus_router( files_dir="./uploads", on_upload_complete=log_upload, ) ``` -------------------------------- ### Create tus router with custom options Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Initialize the tus router with custom parameters including prefix, storage directory, max upload size, and retention period. ```python from tuspyserver import create_tus_router tus_router = create_tus_router( prefix="files", # route prefix (default: 'files') files_dir="/tmp/files", # path to store files max_size=128_849_018_880, # max upload size in bytes (default is ~128GB) auth=noop, # authentication dependency days_to_keep=5, # retention period on_upload_complete=None, # upload callback upload_complete_dep=None, # upload callback (dependency injector) pre_create_hook=None, # pre-creation callback pre_create_dep=None, # pre-creation callback (dependency injector) file_dep=None, # file path callback (dependency injector) ) ``` -------------------------------- ### Initialize Uppy with Tus Endpoint Source: https://github.com/edihasaj/tuspyserver/blob/main/example/frontend/index.html This snippet shows how to initialize Uppy, configure the Dashboard UI, and connect to a TuspyServer endpoint using the Tus protocol. It sets up the upload endpoint and specifies a large chunk size for uploads. ```javascript const url = 'http://localhost:8000' console.log(`Resolved API url: ${url}`) import { Uppy, Dashboard, Tus } from "https://releases.transloadit.com/uppy/v4.13.3/uppy.min.mjs" const uppy = new Uppy() .use(Dashboard, { inline: true, target: '#drag-drop-area' }) .use(Tus, { endpoint: `${url}/files/`, chunkSize: 100_000_000_000, }) uppy.on('complete', (result) => { console.log(result) }) ``` -------------------------------- ### Pre-Create Hook with Dependency Injection for Validation Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Implement a pre-create hook using dependency injection to validate user uploads before creation. This allows for authentication, permission checks, and quota validation. ```python from fastapi import Depends, HTTPException from your_app.dependencies import get_db, get_current_user def validate_user_upload( db=Depends(get_db), current_user=Depends(get_current_user), ) -> Callable[[dict, dict], None]: # callback function async def handler(metadata: dict, upload_info: dict): # Check user permissions if not current_user.can_upload: raise HTTPException(status_code=403, detail="Upload not allowed") # Validate against user's quota user_uploads = await db.get_user_uploads(current_user.id) if len(user_uploads) >= current_user.upload_limit: raise HTTPException(status_code=429, detail="Upload quota exceeded") # Log the upload attempt await db.log_upload_attempt(current_user.id, metadata, upload_info) return handler # Include router with the pre-create DI hook app.include_router( create_tus_router( pre_create_dep=validate_user_upload, ) ) ``` -------------------------------- ### Pre-Create Hook for validation Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Implement a pre-create hook to validate upload metadata, file size, and file type before a file is created on the server. ```python from tuspyserver import create_tus_router from fastapi import HTTPException def validate_upload(metadata: dict, upload_info: dict): # Validate required metadata if "filename" not in metadata: raise HTTPException(status_code=400, detail="Filename is required") # Check file size limits if upload_info["size"] and upload_info["size"] > 100_000_000: # 100MB raise HTTPException(status_code=413, detail="File too large") # Validate file type if "filetype" in metadata: allowed_types = ["image/jpeg", "image/png", "application/pdf"] if metadata["filetype"] not in allowed_types: raise HTTPException(status_code=400, detail="File type not allowed") # Use the hook tus_router = create_tus_router( files_dir="./uploads", pre_create_hook=validate_upload, ) ``` -------------------------------- ### Define Factory Dependency for Upload Logging Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Use dependency injection to create a factory function that logs user uploads. This is useful for performing actions after a file upload is complete. ```python from fastapi import Depends from your_app.dependencies import get_db, get_current_user # factory function def log_user_upload( db=Depends(get_db), current_user=Depends(get_current_user), ) -> Callable[[str, dict], None]: # callback function async def handler(file_path: str, metadata: dict): # perform validation or post-processing await db.log_upload(current_user.id, metadata) await process_file(file_path) return handler # Include router with the DI hook app.include_router( create_api_router( upload_complete_dep=log_user_upload, ) ) ``` -------------------------------- ### Pre-Create Hook Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md A callback function that is executed before a new upload is created. It allows for validation of metadata and authentication before the file is persisted. ```APIDOC ## Pre-Create Hook ### Description Allows validation of metadata and authentication before a file is created on the server. Useful for checking required fields, file types, user permissions, and applying custom business logic. ### Parameters - **metadata** (dict) - A dictionary containing the decoded upload metadata. - **upload_info** (dict) - A dictionary with upload parameters (size, defer_length, expires). ### Example Usage ```python from fastapi import HTTPException def validate_upload(metadata: dict, upload_info: dict): # Validate required metadata if "filename" not in metadata: raise HTTPException(status_code=400, detail="Filename is required") # Check file size limits if upload_info["size"] and upload_info["size"] > 100_000_000: # 100MB raise HTTPException(status_code=413, detail="File too large") # Validate file type if "filetype" in metadata: allowed_types = ["image/jpeg", "image/png", "application/pdf"] if metadata["filetype"] not in allowed_types: raise HTTPException(status_code=400, detail="File type not allowed") # Use the hook when creating the router tus_router = create_tus_router( files_dir="./uploads", pre_create_hook=validate_upload, ) ``` ``` -------------------------------- ### create_tus_router Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Initializes the tus router for a FastAPI application. It allows configuration of various aspects of the tus server, including storage, size limits, authentication, retention, and callbacks for upload events. ```APIDOC ## create_tus_router ### Description Initializes and returns a FastAPI router that implements the tus upload protocol. ### Method `create_tus_router( prefix: str = "files", files_dir: str = "/tmp/files", max_size: int = 128_849_018_880, auth: Callable = noop, days_to_keep: int = 5, on_upload_complete: Optional[Callable] = None, upload_complete_dep: Optional[Callable] = None, pre_create_hook: Optional[Callable] = None, pre_create_dep: Optional[Callable] = None, file_dep: Optional[Callable] = None, ) -> APIRouter ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python from tuspyserver import create_tus_router tus_router = create_tus_router( prefix="files", files_dir="/tmp/files", max_size=128_849_018_880, auth=noop, days_to_keep=5, on_upload_complete=None, upload_complete_dep=None, pre_create_hook=None, pre_create_dep=None, file_dep=None, ) ``` ### Response #### Success Response (200) Returns a FastAPI `APIRouter` instance configured for tus uploads. #### Response Example None ``` -------------------------------- ### On Upload Complete Hook Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md A callback function executed after a file upload is successfully completed. It receives the file path and metadata for further processing. ```APIDOC ## On Upload Complete Hook ### Description A callback function that is executed after a file upload is successfully completed. It receives the path to the uploaded file and its associated metadata. ### Parameters - **file_path** (str) - The absolute path to the completed uploaded file. - **metadata** (dict) - A dictionary containing the metadata associated with the upload. ### Example Usage ```python def log_upload(file_path: str, metadata: dict): print("Upload complete") print(file_path) print(metadata) # Use the hook when including the router app.include_router( create_tus_router( files_dir="./uploads", on_upload_complete=log_upload, ) ) ``` ``` -------------------------------- ### Schedule Expiration File Cleanup Source: https://github.com/edihasaj/tuspyserver/blob/main/README.md Schedule the `remove_expired_files` function to run periodically using APScheduler. This ensures that expired files are automatically cleaned up. ```python from tuspyserver import create_tus_router from apscheduler.schedulers.background import BackgroundScheduler tus_router = create_tus_router( days_to_keep = 23 # configure retention period; defaults to 5 days ) scheduler = BackgroundScheduler() scheduler.add_job( lambda: tus_router.remove_expired_files(), trigger='cron', hour=1, ) scheduler.start() ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.