# FastAPI Timeout Middleware
A configurable timeout middleware for FastAPI applications that automatically handles request timeouts with customizable error responses.
## Features
- ⏱️ **Configurable timeout duration** - Set custom timeout values per application or per endpoint
- 🎯 **Per-endpoint control** - Use `@timeout()` decorator for fine-grained timeout control
- 📝 **Customizable error responses** - Configure status codes, messages, and response format
- 🔧 **Multiple integration methods** - Use as ASGI middleware, HTTP middleware, or endpoint decorator
- 📊 **Processing time tracking** - Optional inclusion of actual processing time in timeout responses
- � **Custom timeout handlers** - Provide your own timeout response logic
- 🚀 **High performance** - Minimal overhead using asyncio
- 📚 **Type hints included** - Full typing support for better IDE integration
- ⚠️ **Safe defaults** - Prevents problematic HTTP 408 status code usage
## Installation
```bash
pip install fastapi-timeout
```
## Quick Start
### Method 1: ASGI Middleware (Recommended)
```python
from fastapi import FastAPI
from fastapi_timeout import TimeoutMiddleware
app = FastAPI()
# Add timeout middleware with 5 second timeout
app.add_middleware(TimeoutMiddleware, timeout_seconds=5.0)
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/slow")
async def slow_endpoint():
import asyncio
await asyncio.sleep(10) # This will timeout after 5 seconds
return {"message": "This will never be reached"}
```
### Method 2: HTTP Middleware Decorator
```python
from fastapi import FastAPI, Request
from fastapi_timeout import timeout_middleware
app = FastAPI()
@app.middleware("http")
async def add_timeout(request: Request, call_next):
timeout_handler = timeout_middleware(timeout_seconds=5.0)
return await timeout_handler(request, call_next)
```
### Method 3: Per-Endpoint Decorator
```python
from fastapi import FastAPI
from fastapi_timeout import timeout
app = FastAPI()
@app.get("/fast-endpoint")
@timeout(5.0) # 5 second timeout for this endpoint only
async def fast_endpoint():
return {"message": "Fast endpoint with 5s timeout"}
@app.get("/slow-endpoint")
@timeout(30.0, timeout_status_code=503, timeout_message="Slow operation timeout")
async def slow_endpoint():
return {"message": "Slow endpoint with 30s timeout"}
```
## Configuration Options
### Basic Configuration
```python
app.add_middleware(
TimeoutMiddleware,
timeout_seconds=10.0, # Timeout after 10 seconds
timeout_status_code=503, # Return 503 Service Unavailable
timeout_message="Request timeout", # Custom error message
include_process_time=True # Include processing time in response
)
```
**⚠️ Important:** Do not use HTTP status code 408 (Request Timeout) as it causes browsers and HTTP clients to automatically retry requests, which can lead to unexpected behavior and increased server load. Use 504 (Gateway Timeout) or 503 (Service Unavailable) instead.
### Advanced Configuration with Custom Handler
```python
from fastapi import Request, Response
from fastapi.responses import JSONResponse
def custom_timeout_handler(request: Request, process_time: float) -> Response:
return JSONResponse(
status_code=503,
content={
"error": "Service temporarily unavailable",
"path": request.url.path,
"method": request.method,
"timeout_duration": process_time,
"retry_after": 60
},
headers={"Retry-After": "60"}
)
app.add_middleware(
TimeoutMiddleware,
timeout_seconds=15.0,
custom_timeout_handler=custom_timeout_handler
)
```
## Per-Endpoint Timeout Control
### Using the @timeout Decorator
For fine-grained control, you can apply different timeouts to specific endpoints using the `@timeout` decorator:
```python
from fastapi import FastAPI, Request
from fastapi_timeout import timeout, TimeoutMiddleware
from fastapi.responses import JSONResponse
app = FastAPI()
# Optional: Global fallback timeout
app.add_middleware(TimeoutMiddleware, timeout_seconds=30.0)
@app.get("/api/quick")
@timeout(2.0) # 2 second timeout
async def quick_operation():
return {"message": "Quick operation"}
@app.get("/api/medium")
@timeout(10.0, timeout_message="Medium operation timed out")
async def medium_operation():
return {"message": "Medium operation"}
@app.get("/api/batch")
@timeout(60.0, timeout_status_code=503)
async def batch_operation():
return {"message": "Long batch operation"}
# Custom timeout handler for specific endpoint
def vip_timeout_handler(request: Request, process_time: float):
return JSONResponse(
status_code=503,
content={
"error": "VIP service busy",
"retry_after": 30,
"processing_time": process_time
}
)
@app.get("/api/vip")
@timeout(5.0, custom_timeout_handler=vip_timeout_handler)
async def vip_endpoint():
return {"message": "VIP operation"}
```
### Decorator Features
- **Per-endpoint control**: Each endpoint can have its own timeout
- **Overrides global middleware**: Decorator timeouts take precedence
- **All configuration options**: Supports custom messages, status codes, and handlers
- **Easy to apply**: Simply add `@timeout(seconds)` below your route decorator
### Mixed Approach
You can combine global middleware with per-endpoint decorators:
```python
# Global timeout for most endpoints
app.add_middleware(TimeoutMiddleware, timeout_seconds=10.0)
# Override for specific endpoints
@app.get("/slow-report")
@timeout(60.0) # This endpoint gets 60 seconds instead of 10
async def generate_report():
return {"message": "Report generated"}
```
## Response Format
### Default Timeout Response
When a request times out, the middleware returns a JSON response:
```json
{
"detail": "Request processing time exceeded limit",
"timeout_seconds": 5.0,
"processing_time": 5.002
}
```
### Customizable Fields
- `timeout_status_code`: HTTP status code (default: 504 Gateway Timeout)
- **⚠️ Avoid 408 (Request Timeout)** - causes automatic retries in browsers/clients
- **Recommended**: 504 (Gateway Timeout) or 503 (Service Unavailable)
- `timeout_message`: Error message (default: "Request processing time exceeded limit")
- `include_process_time`: Whether to include actual processing time (default: True)
## Use Cases
### Different Timeouts for Different Operations
```python
@app.get("/api/search")
@timeout(5.0) # Quick search
async def search(): pass
@app.post("/api/upload")
@timeout(120.0) # File upload
async def upload(): pass
@app.get("/api/report")
@timeout(300.0) # Long report generation
async def generate_report(): pass
```
### Web APIs with Database Queries
```python
# Prevent hanging database queries
app.add_middleware(TimeoutMiddleware, timeout_seconds=30.0)
```
### Microservices with External Dependencies
```python
# Timeout requests that depend on external services
app.add_middleware(
TimeoutMiddleware,
timeout_seconds=10.0,
timeout_status_code=503,
timeout_message="Service temporarily unavailable"
)
```
### File Upload Endpoints
```python
# Different timeout for file upload routes
from starlette.middleware.base import BaseHTTPMiddleware
class ConditionalTimeoutMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
if request.url.path.startswith("/upload"):
# Longer timeout for uploads
timeout_handler = timeout_middleware(timeout_seconds=60.0)
return await timeout_handler(request, call_next)
else:
# Standard timeout for other endpoints
timeout_handler = timeout_middleware(timeout_seconds=5.0)
return await timeout_handler(request, call_next)
app.add_middleware(ConditionalTimeoutMiddleware)
```
## Testing
The package includes comprehensive tests. To run them:
```bash
pip install fastapi-timeout[dev]
pytest
```
## Requirements
- Python 3.7+
- FastAPI 0.65.0+
- Starlette 0.14.0+
## License
MIT License - see LICENSE file for details.
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
Raw data
{
"_id": null,
"home_page": null,
"name": "fastapi-timeout",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": null,
"keywords": "fastapi, timeout, middleware, async",
"author": null,
"author_email": "Zuhair Abdulla <zuhairamahdi@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/74/0a/ec2143dae0d765f13e8ac270db4168e5c16fc9f9d6e542c1e70293ceeda6/fastapi_timeout-0.1.1.post2.tar.gz",
"platform": null,
"description": "# FastAPI Timeout Middleware\n\nA configurable timeout middleware for FastAPI applications that automatically handles request timeouts with customizable error responses.\n\n## Features\n\n- \u23f1\ufe0f **Configurable timeout duration** - Set custom timeout values per application or per endpoint\n- \ud83c\udfaf **Per-endpoint control** - Use `@timeout()` decorator for fine-grained timeout control\n- \ud83d\udcdd **Customizable error responses** - Configure status codes, messages, and response format\n- \ud83d\udd27 **Multiple integration methods** - Use as ASGI middleware, HTTP middleware, or endpoint decorator\n- \ud83d\udcca **Processing time tracking** - Optional inclusion of actual processing time in timeout responses\n- \ufffd **Custom timeout handlers** - Provide your own timeout response logic\n- \ud83d\ude80 **High performance** - Minimal overhead using asyncio\n- \ud83d\udcda **Type hints included** - Full typing support for better IDE integration\n- \u26a0\ufe0f **Safe defaults** - Prevents problematic HTTP 408 status code usage\n\n## Installation\n\n```bash\npip install fastapi-timeout\n```\n\n## Quick Start\n\n### Method 1: ASGI Middleware (Recommended)\n\n```python\nfrom fastapi import FastAPI\nfrom fastapi_timeout import TimeoutMiddleware\n\napp = FastAPI()\n\n# Add timeout middleware with 5 second timeout\napp.add_middleware(TimeoutMiddleware, timeout_seconds=5.0)\n\n@app.get(\"/\")\nasync def root():\n return {\"message\": \"Hello World\"}\n\n@app.get(\"/slow\")\nasync def slow_endpoint():\n import asyncio\n await asyncio.sleep(10) # This will timeout after 5 seconds\n return {\"message\": \"This will never be reached\"}\n```\n\n### Method 2: HTTP Middleware Decorator\n\n```python\nfrom fastapi import FastAPI, Request\nfrom fastapi_timeout import timeout_middleware\n\napp = FastAPI()\n\n@app.middleware(\"http\")\nasync def add_timeout(request: Request, call_next):\n timeout_handler = timeout_middleware(timeout_seconds=5.0)\n return await timeout_handler(request, call_next)\n```\n\n### Method 3: Per-Endpoint Decorator\n\n```python\nfrom fastapi import FastAPI\nfrom fastapi_timeout import timeout\n\napp = FastAPI()\n\n@app.get(\"/fast-endpoint\")\n@timeout(5.0) # 5 second timeout for this endpoint only\nasync def fast_endpoint():\n return {\"message\": \"Fast endpoint with 5s timeout\"}\n\n@app.get(\"/slow-endpoint\") \n@timeout(30.0, timeout_status_code=503, timeout_message=\"Slow operation timeout\")\nasync def slow_endpoint():\n return {\"message\": \"Slow endpoint with 30s timeout\"}\n```\n\n## Configuration Options\n\n### Basic Configuration\n\n```python\napp.add_middleware(\n TimeoutMiddleware,\n timeout_seconds=10.0, # Timeout after 10 seconds\n timeout_status_code=503, # Return 503 Service Unavailable \n timeout_message=\"Request timeout\", # Custom error message\n include_process_time=True # Include processing time in response\n)\n```\n\n**\u26a0\ufe0f Important:** Do not use HTTP status code 408 (Request Timeout) as it causes browsers and HTTP clients to automatically retry requests, which can lead to unexpected behavior and increased server load. Use 504 (Gateway Timeout) or 503 (Service Unavailable) instead.\n\n### Advanced Configuration with Custom Handler\n\n```python\nfrom fastapi import Request, Response\nfrom fastapi.responses import JSONResponse\n\ndef custom_timeout_handler(request: Request, process_time: float) -> Response:\n return JSONResponse(\n status_code=503,\n content={\n \"error\": \"Service temporarily unavailable\",\n \"path\": request.url.path,\n \"method\": request.method,\n \"timeout_duration\": process_time,\n \"retry_after\": 60\n },\n headers={\"Retry-After\": \"60\"}\n )\n\napp.add_middleware(\n TimeoutMiddleware,\n timeout_seconds=15.0,\n custom_timeout_handler=custom_timeout_handler\n)\n```\n\n## Per-Endpoint Timeout Control\n\n### Using the @timeout Decorator\n\nFor fine-grained control, you can apply different timeouts to specific endpoints using the `@timeout` decorator:\n\n```python\nfrom fastapi import FastAPI, Request\nfrom fastapi_timeout import timeout, TimeoutMiddleware\nfrom fastapi.responses import JSONResponse\n\napp = FastAPI()\n\n# Optional: Global fallback timeout\napp.add_middleware(TimeoutMiddleware, timeout_seconds=30.0)\n\n@app.get(\"/api/quick\")\n@timeout(2.0) # 2 second timeout\nasync def quick_operation():\n return {\"message\": \"Quick operation\"}\n\n@app.get(\"/api/medium\")\n@timeout(10.0, timeout_message=\"Medium operation timed out\")\nasync def medium_operation():\n return {\"message\": \"Medium operation\"}\n\n@app.get(\"/api/batch\")\n@timeout(60.0, timeout_status_code=503)\nasync def batch_operation():\n return {\"message\": \"Long batch operation\"}\n\n# Custom timeout handler for specific endpoint\ndef vip_timeout_handler(request: Request, process_time: float):\n return JSONResponse(\n status_code=503,\n content={\n \"error\": \"VIP service busy\",\n \"retry_after\": 30,\n \"processing_time\": process_time\n }\n )\n\n@app.get(\"/api/vip\")\n@timeout(5.0, custom_timeout_handler=vip_timeout_handler)\nasync def vip_endpoint():\n return {\"message\": \"VIP operation\"}\n```\n\n### Decorator Features\n\n- **Per-endpoint control**: Each endpoint can have its own timeout\n- **Overrides global middleware**: Decorator timeouts take precedence\n- **All configuration options**: Supports custom messages, status codes, and handlers\n- **Easy to apply**: Simply add `@timeout(seconds)` below your route decorator\n\n### Mixed Approach\n\nYou can combine global middleware with per-endpoint decorators:\n\n```python\n# Global timeout for most endpoints\napp.add_middleware(TimeoutMiddleware, timeout_seconds=10.0)\n\n# Override for specific endpoints\n@app.get(\"/slow-report\")\n@timeout(60.0) # This endpoint gets 60 seconds instead of 10\nasync def generate_report():\n return {\"message\": \"Report generated\"}\n```\n\n## Response Format\n\n### Default Timeout Response\n\nWhen a request times out, the middleware returns a JSON response:\n\n```json\n{\n \"detail\": \"Request processing time exceeded limit\",\n \"timeout_seconds\": 5.0,\n \"processing_time\": 5.002\n}\n```\n\n### Customizable Fields\n\n- `timeout_status_code`: HTTP status code (default: 504 Gateway Timeout)\n - **\u26a0\ufe0f Avoid 408 (Request Timeout)** - causes automatic retries in browsers/clients\n - **Recommended**: 504 (Gateway Timeout) or 503 (Service Unavailable)\n- `timeout_message`: Error message (default: \"Request processing time exceeded limit\")\n- `include_process_time`: Whether to include actual processing time (default: True)\n\n## Use Cases\n\n### Different Timeouts for Different Operations\n```python\n@app.get(\"/api/search\")\n@timeout(5.0) # Quick search\nasync def search(): pass\n\n@app.post(\"/api/upload\")\n@timeout(120.0) # File upload\nasync def upload(): pass\n\n@app.get(\"/api/report\")\n@timeout(300.0) # Long report generation\nasync def generate_report(): pass\n```\n\n### Web APIs with Database Queries\n```python\n# Prevent hanging database queries\napp.add_middleware(TimeoutMiddleware, timeout_seconds=30.0)\n```\n\n### Microservices with External Dependencies\n```python\n# Timeout requests that depend on external services\napp.add_middleware(\n TimeoutMiddleware, \n timeout_seconds=10.0,\n timeout_status_code=503,\n timeout_message=\"Service temporarily unavailable\"\n)\n```\n\n### File Upload Endpoints\n```python\n# Different timeout for file upload routes\nfrom starlette.middleware.base import BaseHTTPMiddleware\n\nclass ConditionalTimeoutMiddleware(BaseHTTPMiddleware):\n async def dispatch(self, request, call_next):\n if request.url.path.startswith(\"/upload\"):\n # Longer timeout for uploads\n timeout_handler = timeout_middleware(timeout_seconds=60.0)\n return await timeout_handler(request, call_next)\n else:\n # Standard timeout for other endpoints\n timeout_handler = timeout_middleware(timeout_seconds=5.0)\n return await timeout_handler(request, call_next)\n\napp.add_middleware(ConditionalTimeoutMiddleware)\n```\n\n## Testing\n\nThe package includes comprehensive tests. To run them:\n\n```bash\npip install fastapi-timeout[dev]\npytest\n```\n\n## Requirements\n\n- Python 3.7+\n- FastAPI 0.65.0+\n- Starlette 0.14.0+\n\n## License\n\nMIT License - see LICENSE file for details.\n\n## Contributing\n\nContributions are welcome! Please feel free to submit a Pull Request.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Configurable timeout middleware for FastAPI applications",
"version": "0.1.1.post2",
"project_urls": {
"Bug Tracker": "https://github.com/zuhairamahdi/fastapi-timeout/issues",
"Documentation": "https://github.com/zuhairamahdi/fastapi-timeout#readme",
"Homepage": "https://github.com/zuhairamahdi/fastapi-timeout",
"Repository": "https://github.com/zuhairamahdi/fastapi-timeout"
},
"split_keywords": [
"fastapi",
" timeout",
" middleware",
" async"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "563239d95b47736354de4b8d1ac574cd1737db07a1d4c65fab2402ea64a21853",
"md5": "2d69a23a0553b10d5667b400974dc147",
"sha256": "d0ec682d694135d041eddc00b8d0f94bcb0918aee8803c440c4d669bff1a94e2"
},
"downloads": -1,
"filename": "fastapi_timeout-0.1.1.post2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2d69a23a0553b10d5667b400974dc147",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 8263,
"upload_time": "2025-09-15T17:24:38",
"upload_time_iso_8601": "2025-09-15T17:24:38.287336Z",
"url": "https://files.pythonhosted.org/packages/56/32/39d95b47736354de4b8d1ac574cd1737db07a1d4c65fab2402ea64a21853/fastapi_timeout-0.1.1.post2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "740aec2143dae0d765f13e8ac270db4168e5c16fc9f9d6e542c1e70293ceeda6",
"md5": "414fff2985bcac2d40b26fedd04dff92",
"sha256": "eefcc58bc8b9b3a3c1e22e4a3b1e7d7baff6a9f067ce1c2607699d55cbb19f91"
},
"downloads": -1,
"filename": "fastapi_timeout-0.1.1.post2.tar.gz",
"has_sig": false,
"md5_digest": "414fff2985bcac2d40b26fedd04dff92",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 12091,
"upload_time": "2025-09-15T17:24:39",
"upload_time_iso_8601": "2025-09-15T17:24:39.816469Z",
"url": "https://files.pythonhosted.org/packages/74/0a/ec2143dae0d765f13e8ac270db4168e5c16fc9f9d6e542c1e70293ceeda6/fastapi_timeout-0.1.1.post2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-15 17:24:39",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "zuhairamahdi",
"github_project": "fastapi-timeout",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "fastapi-timeout"
}
JSON
