# kusho-capture
Kusho-capture is a lightweight HTTP traffic capture middleware for Python web applications. It seamlessly integrates with popular frameworks like FastAPI, Flask, and Django to record API traffic for testing, debugging, and monitoring purposes.
## Features
- 🔄 Automatic HTTP traffic capture for API endpoints
- 🎯 Configurable URL pattern matching
- 📦 Support for both WSGI and ASGI applications
- 🔍 Detailed request/response logging
- ⚡ Async support for modern web frameworks
- 🎛️ Configurable sampling rate for high-traffic applications
- 📊 Batch processing of captured events
- 🚀 Framework auto-detection
## Installation
```bash
pip install kusho-capture
```
For framework-specific dependencies:
```bash
pip install kusho-capture[fastapi] # For FastAPI support
pip install kusho-capture[flask] # For Flask support
pip install kusho-capture[django] # For Django support
```
## Quick Start
### FastAPI Example
```python
from fastapi import FastAPI
from kusho_capture import EventCollector, setup_traffic_capture
app = FastAPI()
collector = EventCollector(
collector_url="https://your-collector-endpoint.com",
sample_rate=0.1,
batch_size=100
)
app = setup_traffic_capture(app, collector, framework="fastapi")
@app.get("/api/items")
async def get_items():
return {"items": ["item1", "item2"]}
```
### Flask Example
```python
from flask import Flask
from kusho_capture import EventCollector, setup_traffic_capture
app = Flask(__name__)
collector = EventCollector(
collector_url="https://your-collector-endpoint.com",
sample_rate=0.1,
batch_size=100
)
app.wsgi_app = setup_traffic_capture(app.wsgi_app, collector, framework="flask")
@app.route("/api/items")
def get_items():
return {"items": ["item1", "item2"]}
```
### Django Example
```python
# settings.py
MIDDLEWARE = [
'kusho_capture.WSGIMiddleware',
# ... other middleware
]
# somewhere in your configuration
from kusho_capture import EventCollector, setup_traffic_capture
collector = EventCollector(
collector_url="https://your-collector-endpoint.com",
sample_rate=0.1,
batch_size=100
)
application = setup_traffic_capture(application, collector, framework="django")
```
## Configuration
### EventCollector Options
- `collector_url`: URL of your event collection endpoint
- `batch_size`: Number of events to batch before sending (default: 100)
- `flush_interval`: Maximum time to wait before sending a batch in seconds (default: 60)
- `max_queue_size`: Maximum number of events to queue (default: 10000)
- `sample_rate`: Percentage of requests to capture (default: 0.1 = 10%)
### Middleware Options
- `url_patterns`: List of URL patterns to match for capture (default: ['/api/'])
- `framework`: Auto-detected by default, can be explicitly set to 'fastapi', 'flask', or 'django'
## Event Data Structure
Captured events include:
- Timestamp
- Request path and method
- Request headers
- Query parameters
- Request body (for POST/PUT/PATCH requests)
- Response status and headers
- Response time
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
## License
This project is licensed under the MIT License - see the LICENSE file for details.
Raw data
{
"_id": null,
"home_page": "https://github.com/kusho-co/kusho-capture",
"name": "kusho-capture",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": null,
"keywords": "http, middleware, traffic capture, api testing, monitoring, debugging",
"author": "KushoAI",
"author_email": "support@kusho.co",
"download_url": "https://files.pythonhosted.org/packages/1a/35/411bb1fa50a3cbfd2bca1838cceed1856135003ef81b203ddabbb9db8b84/kusho_capture-0.1.0.tar.gz",
"platform": null,
"description": "# kusho-capture\n\nKusho-capture is a lightweight HTTP traffic capture middleware for Python web applications. It seamlessly integrates with popular frameworks like FastAPI, Flask, and Django to record API traffic for testing, debugging, and monitoring purposes.\n\n## Features\n\n- \ud83d\udd04 Automatic HTTP traffic capture for API endpoints\n- \ud83c\udfaf Configurable URL pattern matching\n- \ud83d\udce6 Support for both WSGI and ASGI applications\n- \ud83d\udd0d Detailed request/response logging\n- \u26a1 Async support for modern web frameworks\n- \ud83c\udf9b\ufe0f Configurable sampling rate for high-traffic applications\n- \ud83d\udcca Batch processing of captured events\n- \ud83d\ude80 Framework auto-detection\n\n## Installation\n\n```bash\npip install kusho-capture\n```\n\nFor framework-specific dependencies:\n\n```bash\npip install kusho-capture[fastapi] # For FastAPI support\npip install kusho-capture[flask] # For Flask support\npip install kusho-capture[django] # For Django support\n```\n\n## Quick Start\n\n### FastAPI Example\n\n```python\nfrom fastapi import FastAPI\nfrom kusho_capture import EventCollector, setup_traffic_capture\n\napp = FastAPI()\ncollector = EventCollector(\n collector_url=\"https://your-collector-endpoint.com\",\n sample_rate=0.1,\n batch_size=100\n)\n\napp = setup_traffic_capture(app, collector, framework=\"fastapi\")\n\n@app.get(\"/api/items\")\nasync def get_items():\n return {\"items\": [\"item1\", \"item2\"]}\n```\n\n### Flask Example\n\n```python\nfrom flask import Flask\nfrom kusho_capture import EventCollector, setup_traffic_capture\n\napp = Flask(__name__)\ncollector = EventCollector(\n collector_url=\"https://your-collector-endpoint.com\",\n sample_rate=0.1,\n batch_size=100\n)\n\napp.wsgi_app = setup_traffic_capture(app.wsgi_app, collector, framework=\"flask\")\n\n@app.route(\"/api/items\")\ndef get_items():\n return {\"items\": [\"item1\", \"item2\"]}\n```\n\n### Django Example\n\n```python\n# settings.py\nMIDDLEWARE = [\n 'kusho_capture.WSGIMiddleware',\n # ... other middleware\n]\n\n# somewhere in your configuration\nfrom kusho_capture import EventCollector, setup_traffic_capture\n\ncollector = EventCollector(\n collector_url=\"https://your-collector-endpoint.com\",\n sample_rate=0.1,\n batch_size=100\n)\n\napplication = setup_traffic_capture(application, collector, framework=\"django\")\n```\n\n## Configuration\n\n### EventCollector Options\n\n- `collector_url`: URL of your event collection endpoint\n- `batch_size`: Number of events to batch before sending (default: 100)\n- `flush_interval`: Maximum time to wait before sending a batch in seconds (default: 60)\n- `max_queue_size`: Maximum number of events to queue (default: 10000)\n- `sample_rate`: Percentage of requests to capture (default: 0.1 = 10%)\n\n### Middleware Options\n\n- `url_patterns`: List of URL patterns to match for capture (default: ['/api/'])\n- `framework`: Auto-detected by default, can be explicitly set to 'fastapi', 'flask', or 'django'\n\n## Event Data Structure\n\nCaptured events include:\n- Timestamp\n- Request path and method\n- Request headers\n- Query parameters\n- Request body (for POST/PUT/PATCH requests)\n- Response status and headers\n- Response time\n\n## Contributing\n\nContributions are welcome! Please feel free to submit a Pull Request.\n\n## License\n\nThis project is licensed under the MIT License - see the LICENSE file for details.\n",
"bugtrack_url": null,
"license": null,
"summary": "HTTP traffic capture middleware for Python web applications",
"version": "0.1.0",
"project_urls": {
"Homepage": "https://github.com/kusho-co/kusho-capture"
},
"split_keywords": [
"http",
" middleware",
" traffic capture",
" api testing",
" monitoring",
" debugging"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "cad3c34db2dd25823e8dc0c2c78d21fe0247c76fc942b1d72bdd70f8743efcc0",
"md5": "def0fbc4df664f13db4355adae723692",
"sha256": "396748912921fb95b9d9bb9f9f9c3c09cc517a1852d893096a86f5619f4061c7"
},
"downloads": -1,
"filename": "kusho_capture-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "def0fbc4df664f13db4355adae723692",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 6685,
"upload_time": "2024-12-14T19:43:54",
"upload_time_iso_8601": "2024-12-14T19:43:54.577791Z",
"url": "https://files.pythonhosted.org/packages/ca/d3/c34db2dd25823e8dc0c2c78d21fe0247c76fc942b1d72bdd70f8743efcc0/kusho_capture-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "1a35411bb1fa50a3cbfd2bca1838cceed1856135003ef81b203ddabbb9db8b84",
"md5": "2f9bccfbf57c42351a23efeb4a998ffd",
"sha256": "7d7e925e781d3e3cbb17ef0a2d67397aeebb5cba37f4427a0efc865940d046c4"
},
"downloads": -1,
"filename": "kusho_capture-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "2f9bccfbf57c42351a23efeb4a998ffd",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 6281,
"upload_time": "2024-12-14T19:43:55",
"upload_time_iso_8601": "2024-12-14T19:43:55.923394Z",
"url": "https://files.pythonhosted.org/packages/1a/35/411bb1fa50a3cbfd2bca1838cceed1856135003ef81b203ddabbb9db8b84/kusho_capture-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-14 19:43:55",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "kusho-co",
"github_project": "kusho-capture",
"github_not_found": true,
"lcname": "kusho-capture"
}
JSON
