Imagine you are a Data Analyst or a DevOps Engineer responsible for processing massive 1GB log files or converting complex CSV reports into JSON. If you trigger this via a standard web request, the connection will likely time out before the work finishes. It is like standing at a coffee shop counter and waiting for the beans to be roasted, ground, and brewed while a line forms behind you. Instead, you want a system where you hand over the file, receive a **Task ID**, and walk away. This script creates an automated **Background Worker** that handles heavy lifting while keeping the API responsive.<\/p>\n
We leverage the **FastAPI** framework and its **BackgroundTasks** utility. This allows the server to return an immediate success response to the user while continuing to execute the processing logic in a separate thread. We use **pathlib** for robust file system management and **uuid** to ensure every task has a unique identity.<\/p>\n
Ensure you have Python 3.9 or higher installed. Install the required dependencies using the terminal:<\/p>\n
\n\"\"\"\n-----------------------------------------------------------------------\nAuthors: Sharanam & Vaishali Shah\nRecipe: FastAPI Background File Processor\nIntent: Automate long-running file tasks with non-blocking status tracking.\n-----------------------------------------------------------------------\n\"\"\"\n\nimport time\nimport shutil\nfrom pathlib import Path\nfrom uuid import uuid4\nfrom typing import Dict\nfrom fastapi import FastAPI, BackgroundTasks, UploadFile, File\n\napp = FastAPI()\n\n# Define local directories using pathlib\nBASE_DIR = Path(__file__).resolve().parent\nUPLOAD_DIR = BASE_DIR \/ **uploads**\nRESULT_DIR = BASE_DIR \/ **results**\n\n# Ensure directories exist\nUPLOAD_DIR.mkdir(exist_ok=True)\nRESULT_DIR.mkdir(exist_ok=True)\n\n# In-memory database to track task status\ntasks_db: Dict[str, str] = {}\n\ndef process_file_task(task_id: str, file_path: Path):\n \"\"\"\n Simulates a heavy processing task like data cleaning or conversion.\n \"\"\"\n tasks_db[task_id] = **Processing**\n \n try:\n # Simulate a long-running operation (e.g., 10 seconds)\n time.sleep(10)\n \n # Define the output path\n output_file = RESULT_DIR \/ f\"processed_{task_id}.txt\"\n \n # Read the uploaded file and write 'processed' content\n content = file_path.read_text()\n output_file.write_text(f\"PROCESSED CONTENT:n{content.upper()}\")\n \n # Update status\n tasks_db[task_id] = **Completed**\n except Exception as e:\n tasks_db[task_id] = f\"Failed: {str(e)}\"\n\n@app.post(**\/upload**)\nasync def upload_file(background_tasks: BackgroundTasks, file: UploadFile = File(...)):\n # Generate a unique identity for the task\n task_id = str(uuid4())\n file_extension = Path(file.filename).suffix\n local_path = UPLOAD_DIR \/ f\"{task_id}{file_extension}\"\n \n # Save the uploaded file to disk\n with local_path.open(**wb**) as buffer:\n shutil.copyfileobj(file.file, buffer)\n \n # Register the task to run in the background\n tasks_db[task_id] = **Pending**\n background_tasks.add_task(process_file_task, task_id, local_path)\n \n return {\n **message**: **File uploaded successfully**,\n **task_id**: task_id,\n **status_endpoint**: f\"\/status\/{task_id}\"\n }\n\n@app.get(**\/status\/{task_id}**)\nasync def get_status(task_id: str):\n # Retrieve status from the in-memory dictionary\n status = tasks_db.get(task_id, **Not Found**)\n return {**task_id**: task_id, **status**: status}\n\nif __name__ == **__main__**:\n import uvicorn\n uvicorn.run(app, host=**127.0.0.1**, port=8000)\n<\/code><\/pre>\nCode walkthrough<\/h2>\n
The script begins by setting up two directories: **uploads** for incoming files and **results** for processed output. We use **pathlib** to ensure this works seamlessly across Windows, Mac, and Linux. The **tasks_db** dictionary acts as a volatile database to keep track of whether a job is **Pending**, **Processing**, or **Completed**.<\/p>\n
When you hit the **upload** endpoint, the script saves your file and generates a **UUID**. Instead of processing the file immediately, it calls **background_tasks.add_task**. This tells FastAPI to return the JSON response to the user immediately while the **process_file_task** function runs in the background. The **status** endpoint allows the user to poll for updates until the work is done.<\/p>\n
Sample output<\/h2>\n
After running the script with **uvicorn**, you can test it using **curl** or Postman. When you upload a file, the terminal logs the request, and you receive this response:<\/p>\n
\n{\n \"message\": \"File uploaded successfully\",\n \"task_id\": \"550e8400-e29b-41d4-a716-446655440000\",\n \"status_endpoint\": \"\/status\/550e8400-e29b-41d4-a716-446655440000\"\n}\n<\/code><\/pre>\nChecking the status immediately after gives:<\/p>\n
\n{\n \"task_id\": \"550e8400-e29b-41d4-a716-446655440000\",\n \"status\": \"Processing\"\n}\n<\/code><\/pre>\nConclusion<\/h2>\n
You have built a professional-grade asynchronous file processor. This pattern is essential for any modern application that handles data engineering or system automation. By separating the upload from the processing logic, you ensure your application remains fast, stable, and user-friendly even under heavy computational loads.<\/p>\n
\n\n\ud83d\ude80 Don’t Just Learn FastAPI \u2014 Master It.<\/h3>\n
\n This tutorial was just the tip of the iceberg. To truly advance your career and build professional-grade systems<\/strong>, you need the full architectural blueprint.\n <\/p>\n