# Flask HTTP Middleware
[![Downloads](https://static.pepy.tech/personalized-badge/flask-http-middleware?period=total&units=international_system&left_color=black&right_color=orange&left_text=Downloads)](https://pepy.tech/project/flask-http-middleware)
## Installation
```
pip install flask-http-middleware
```
## Description
A module to create flask middleware with direct access to `request` and `response`.
This module implement the starlette's (FastAPI) BaseHTTPMiddleware style to Flask.
## Changelogs
- v0.0
- First Upload
- v0.1
- Allow middlewares stacking
- v0.2
- Add support for `flask>=2.2.x`
- v0.3
- Add support for `flask>=2.3.x`
- v0.4
- Add support for `flask>=3.x`
## How to use ?
### Example: adding a response header
```
import time
from flask import Flask
from flask_http_middleware import MiddlewareManager, BaseHTTPMiddleware
app = Flask(__name__)
class MetricsMiddleware(BaseHTTPMiddleware):
def __init__(self):
super().__init__()
def dispatch(self, request, call_next):
t0 = time.time()
response = call_next(request)
response_time = time.time()-t0
response.headers.add("response_time", response_time)
return response
app.wsgi_app = MiddlewareManager(app)
app.wsgi_app.add_middleware(MetricsMiddleware)
@app.get("/health")
def health():
return {"message":"I'm healthy"}
if __name__ == "__main__":
app.run()
```
- Note: you can put your `MetricsMiddleware` class in different file
Above example is equals with `app.before_request` and `app.after_request` decorated function.
```
@app.before_request
def start_metrics():
g.t0 = time.time()
@app.after_request
def stop_metrics(response):
response_time = time.time()-g.t0
response.headers.add("response_time", response_time)
return response
```
---
### Example: Authentication
```
import time
from flask import Flask, jsonify
from flask_http_middleware import MiddlewareManager, BaseHTTPMiddleware
app = Flask(__name__)
class AccessMiddleware(BaseHTTPMiddleware):
def __init__(self):
super().__init__()
def dispatch(self, request, call_next):
if request.headers.get("token") == "secret":
return call_next(request)
else:
return jsonify({"message":"invalid token"})
app.wsgi_app = MiddlewareManager(app)
app.wsgi_app.add_middleware(AccessMiddleware)
@app.get("/health")
def health():
return {"message":"I'm healthy"}
if __name__ == "__main__":
app.run()
```
---
### Example: add some routers security
```
import time
from flask import Flask, jsonify
from flask_http_middleware import MiddlewareManager, BaseHTTPMiddleware
app = Flask(__name__)
class SecureRoutersMiddleware(BaseHTTPMiddleware):
def __init__(self, secured_routers = []):
super().__init__()
self.secured_routers = secured_routers
def dispatch(self, request, call_next):
if request.path in self.secured_routers:
if request.headers.get("token") == "secret":
return call_next(request)
else:
return jsonify({"message":"invalid token"})
else:
return call_next(request)
secured_routers = ["/check_secured"]
app.wsgi_app = MiddlewareManager(app)
app.wsgi_app.add_middleware(SecureRoutersMiddleware, secured_routers=secured_routers)
@app.get("/health")
def health():
return {"message":"I'm healthy"}
@app.get("/check_secured")
def health():
return {"message":"Security bypassed"}
if __name__ == "__main__":
app.run()
```
---
### Example: add error handling
```
import time
from flask import Flask, jsonify
from flask_http_middleware import MiddlewareManager, BaseHTTPMiddleware
app = Flask(__name__)
class AccessMiddleware(BaseHTTPMiddleware):
def __init__(self):
super().__init__()
def dispatch(self, request, call_next):
if request.headers.get("token") == "secret":
return call_next(request)
else:
raise Exception("Authentication Failed")
def error_handler(self, error):
return jsonify({"error": str(error)})
app.wsgi_app = MiddlewareManager(app)
app.wsgi_app.add_middleware(AccessMiddleware)
@app.get("/health")
def health():
return {"message":"I'm healthy"}
if __name__ == "__main__":
app.run()
```
### Example: Stacking Middleware
You can also stack your middleware
`middleware.py`
```
import time
from flask import jsonify
from flask_http_middleware import BaseHTTPMiddleware
class AccessMiddleware(BaseHTTPMiddleware):
def __init__(self):
super().__init__()
def dispatch(self, request, call_next):
if request.headers.get("token") == "secret":
return call_next(request)
else:
raise Exception("Authentication Failed")
def error_handler(self, error):
return jsonify({"error": str(error)})
class MetricsMiddleware(BaseHTTPMiddleware):
def __init__(self):
super().__init__()
def dispatch(self, request, call_next):
t0 = time.time()
response = call_next(request)
response_time = time.time()-t0
response.headers.add("response_time", response_time)
return response
class SecureRoutersMiddleware(BaseHTTPMiddleware):
def __init__(self, secured_routers = []):
super().__init__()
self.secured_routers = secured_routers
def dispatch(self, request, call_next):
if request.path in self.secured_routers:
if request.headers.get("token") == "secret":
return call_next(request)
else:
return jsonify({"message":"invalid token"})
else:
return call_next(request)
```
your `main.py`
```
import time
from flask import Flask, jsonify
from flask_http_middleware import MiddlewareManager
from middleware import AccessMiddleware, MetricsMiddleware, SecureRoutersMiddleware
app = Flask(__name__)
my_secured_routers = ["/check_secured"]
app.wsgi_app = MiddlewareManager(app)
app.wsgi_app.add_middleware(AccessMiddleware)
app.wsgi_app.add_middleware(MetricsMiddleware)
app.wsgi_app.add_middleware(SecureRoutersMiddleware, secured_routers=my_secured_routers)
@app.get("/health")
def health():
return {"message":"I'm healthy"}
@app.get("/check_secured")
def health():
return {"message":"Security bypassed"}
if __name__ == "__main__":
app.run()
```
Raw data
{
"_id": null,
"home_page": "https://github.com/Danangjoyoo/flask-http-middleware",
"name": "flask-http-middleware",
"maintainer": null,
"docs_url": null,
"requires_python": null,
"maintainer_email": null,
"keywords": "flask, middleware, http, request, response",
"author": "danangjoyoo (Agus Danangjoyo)",
"author_email": "<agus.danangjoyo.blog@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/b8/4d/a953ea317c4819c6cf0b3d3f28848e0ffa158ea8681ab47862f41c7d27d1/flask_http_middleware-0.4.4.tar.gz",
"platform": null,
"description": "# Flask HTTP Middleware\n[![Downloads](https://static.pepy.tech/personalized-badge/flask-http-middleware?period=total&units=international_system&left_color=black&right_color=orange&left_text=Downloads)](https://pepy.tech/project/flask-http-middleware)\n\n\n## Installation\n```\npip install flask-http-middleware\n```\n\n## Description\nA module to create flask middleware with direct access to `request` and `response`.\n\nThis module implement the starlette's (FastAPI) BaseHTTPMiddleware style to Flask.\n\n## Changelogs\n- v0.0\n - First Upload\n- v0.1\n - Allow middlewares stacking\n- v0.2\n - Add support for `flask>=2.2.x`\n- v0.3\n - Add support for `flask>=2.3.x`\n- v0.4\n - Add support for `flask>=3.x`\n\n## How to use ?\n\n### Example: adding a response header\n```\nimport time\nfrom flask import Flask\nfrom flask_http_middleware import MiddlewareManager, BaseHTTPMiddleware\n\napp = Flask(__name__)\n\nclass MetricsMiddleware(BaseHTTPMiddleware):\n def __init__(self):\n super().__init__()\n\n def dispatch(self, request, call_next):\n t0 = time.time()\n response = call_next(request)\n response_time = time.time()-t0\n response.headers.add(\"response_time\", response_time)\n return response\n\napp.wsgi_app = MiddlewareManager(app)\napp.wsgi_app.add_middleware(MetricsMiddleware)\n\n@app.get(\"/health\")\ndef health():\n return {\"message\":\"I'm healthy\"}\n\nif __name__ == \"__main__\":\n app.run()\n```\n- Note: you can put your `MetricsMiddleware` class in different file\n\nAbove example is equals with `app.before_request` and `app.after_request` decorated function.\n\n```\n@app.before_request\ndef start_metrics():\n g.t0 = time.time()\n\n@app.after_request\ndef stop_metrics(response):\n response_time = time.time()-g.t0\n response.headers.add(\"response_time\", response_time)\n return response\n```\n\n---\n\n### Example: Authentication\n```\nimport time\nfrom flask import Flask, jsonify\nfrom flask_http_middleware import MiddlewareManager, BaseHTTPMiddleware\n\napp = Flask(__name__)\n\nclass AccessMiddleware(BaseHTTPMiddleware):\n def __init__(self):\n super().__init__()\n\n def dispatch(self, request, call_next):\n if request.headers.get(\"token\") == \"secret\":\n return call_next(request)\n else:\n return jsonify({\"message\":\"invalid token\"})\n\napp.wsgi_app = MiddlewareManager(app)\napp.wsgi_app.add_middleware(AccessMiddleware)\n\n@app.get(\"/health\")\ndef health():\n return {\"message\":\"I'm healthy\"}\n\nif __name__ == \"__main__\":\n app.run()\n```\n\n---\n\n### Example: add some routers security\n```\nimport time\nfrom flask import Flask, jsonify\nfrom flask_http_middleware import MiddlewareManager, BaseHTTPMiddleware\n\napp = Flask(__name__)\n\nclass SecureRoutersMiddleware(BaseHTTPMiddleware):\n def __init__(self, secured_routers = []):\n super().__init__()\n self.secured_routers = secured_routers\n\n def dispatch(self, request, call_next):\n if request.path in self.secured_routers:\n if request.headers.get(\"token\") == \"secret\":\n return call_next(request)\n else:\n return jsonify({\"message\":\"invalid token\"})\n else:\n return call_next(request)\n\nsecured_routers = [\"/check_secured\"]\n\napp.wsgi_app = MiddlewareManager(app)\napp.wsgi_app.add_middleware(SecureRoutersMiddleware, secured_routers=secured_routers)\n\n@app.get(\"/health\")\ndef health():\n return {\"message\":\"I'm healthy\"}\n\n@app.get(\"/check_secured\")\ndef health():\n return {\"message\":\"Security bypassed\"}\n\nif __name__ == \"__main__\":\n app.run()\n```\n\n---\n\n### Example: add error handling\n```\nimport time\nfrom flask import Flask, jsonify\nfrom flask_http_middleware import MiddlewareManager, BaseHTTPMiddleware\n\napp = Flask(__name__)\n\nclass AccessMiddleware(BaseHTTPMiddleware):\n def __init__(self):\n super().__init__()\n\n def dispatch(self, request, call_next):\n if request.headers.get(\"token\") == \"secret\":\n return call_next(request)\n else:\n raise Exception(\"Authentication Failed\")\n\n def error_handler(self, error):\n return jsonify({\"error\": str(error)})\n\napp.wsgi_app = MiddlewareManager(app)\napp.wsgi_app.add_middleware(AccessMiddleware)\n\n@app.get(\"/health\")\ndef health():\n return {\"message\":\"I'm healthy\"}\n\nif __name__ == \"__main__\":\n app.run()\n```\n\n### Example: Stacking Middleware\nYou can also stack your middleware\n\n`middleware.py`\n```\nimport time\nfrom flask import jsonify\nfrom flask_http_middleware import BaseHTTPMiddleware\n\nclass AccessMiddleware(BaseHTTPMiddleware):\n def __init__(self):\n super().__init__()\n\n def dispatch(self, request, call_next):\n if request.headers.get(\"token\") == \"secret\":\n return call_next(request)\n else:\n raise Exception(\"Authentication Failed\")\n\n def error_handler(self, error):\n return jsonify({\"error\": str(error)})\n\n\nclass MetricsMiddleware(BaseHTTPMiddleware):\n def __init__(self):\n super().__init__()\n\n def dispatch(self, request, call_next):\n t0 = time.time()\n response = call_next(request)\n response_time = time.time()-t0\n response.headers.add(\"response_time\", response_time)\n return response\n\n\nclass SecureRoutersMiddleware(BaseHTTPMiddleware):\n def __init__(self, secured_routers = []):\n super().__init__()\n self.secured_routers = secured_routers\n\n def dispatch(self, request, call_next):\n if request.path in self.secured_routers:\n if request.headers.get(\"token\") == \"secret\":\n return call_next(request)\n else:\n return jsonify({\"message\":\"invalid token\"})\n else:\n return call_next(request)\n\n```\n\nyour `main.py`\n```\nimport time\nfrom flask import Flask, jsonify\nfrom flask_http_middleware import MiddlewareManager\nfrom middleware import AccessMiddleware, MetricsMiddleware, SecureRoutersMiddleware\n\napp = Flask(__name__)\n\nmy_secured_routers = [\"/check_secured\"]\n\napp.wsgi_app = MiddlewareManager(app)\n\napp.wsgi_app.add_middleware(AccessMiddleware)\napp.wsgi_app.add_middleware(MetricsMiddleware)\napp.wsgi_app.add_middleware(SecureRoutersMiddleware, secured_routers=my_secured_routers)\n\n@app.get(\"/health\")\ndef health():\n return {\"message\":\"I'm healthy\"}\n\n@app.get(\"/check_secured\")\ndef health():\n return {\"message\":\"Security bypassed\"}\n\nif __name__ == \"__main__\":\n app.run()\n```\n",
"bugtrack_url": null,
"license": null,
"summary": "A module to create middleware with direct access to `request` and `response`",
"version": "0.4.4",
"project_urls": {
"Homepage": "https://github.com/Danangjoyoo/flask-http-middleware"
},
"split_keywords": [
"flask",
" middleware",
" http",
" request",
" response"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "5f8503c84edec6a13aed5af5af3c2d5acc46c52757375d9acfb1e96b1b3c144c",
"md5": "0c62d4d950f5e5be4707e1d20a86d7e8",
"sha256": "691348d5a04e15cc5178108178a8febe60d242e4fb9ee79832add713de2a9954"
},
"downloads": -1,
"filename": "flask_http_middleware-0.4.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0c62d4d950f5e5be4707e1d20a86d7e8",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": null,
"size": 6157,
"upload_time": "2024-12-09T23:04:31",
"upload_time_iso_8601": "2024-12-09T23:04:31.435399Z",
"url": "https://files.pythonhosted.org/packages/5f/85/03c84edec6a13aed5af5af3c2d5acc46c52757375d9acfb1e96b1b3c144c/flask_http_middleware-0.4.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "b84da953ea317c4819c6cf0b3d3f28848e0ffa158ea8681ab47862f41c7d27d1",
"md5": "0bfaedc8f3769904938c558c8964add0",
"sha256": "415889cece1ce836e47b02ca8b440a0c93fdc63dc1d54e84c2ce9d78799777a1"
},
"downloads": -1,
"filename": "flask_http_middleware-0.4.4.tar.gz",
"has_sig": false,
"md5_digest": "0bfaedc8f3769904938c558c8964add0",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 5820,
"upload_time": "2024-12-09T23:04:33",
"upload_time_iso_8601": "2024-12-09T23:04:33.476678Z",
"url": "https://files.pythonhosted.org/packages/b8/4d/a953ea317c4819c6cf0b3d3f28848e0ffa158ea8681ab47862f41c7d27d1/flask_http_middleware-0.4.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-09 23:04:33",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Danangjoyoo",
"github_project": "flask-http-middleware",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "flask-http-middleware"
}