flask-http-middleware


Nameflask-http-middleware JSON
Version 0.4.1 PyPI version JSON
download
home_pagehttps://github.com/Danangjoyoo/flask-http-middleware
SummaryA module to create middleware with direct access to `request` and `response`
upload_time2023-11-03 04:05:01
maintainer
docs_urlNone
authordanangjoyoo (Agus Danangjoyo)
requires_python
license
keywords flask middleware http request response
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls. 
            # 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": "",
    "docs_url": null,
    "requires_python": "",
    "maintainer_email": "",
    "keywords": "flask,middleware,http,request,response",
    "author": "danangjoyoo (Agus Danangjoyo)",
    "author_email": "<agus.danangjoyo.blog@gmail.com>",
    "download_url": "https://files.pythonhosted.org/packages/07/c6/4fbeb83acaadd5e67ca2ab504b937a52848f160464263491736d4f599bb3/flask-http-middleware-0.4.1.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": "",
    "summary": "A module to create middleware with direct access to `request` and `response`",
    "version": "0.4.1",
    "project_urls": {
        "Homepage": "https://github.com/Danangjoyoo/flask-http-middleware"
    },
    "split_keywords": [
        "flask",
        "middleware",
        "http",
        "request",
        "response"
    ],
    "urls": [
        {
            "comment_text": "",
            "digests": {
                "blake2b_256": "845a9222291851feb404a998c2213329dadf2f70f440f3312b89385b9dfd1bfc",
                "md5": "1f4dd9b893cefe7b2fdfc75c08d8fd5b",
                "sha256": "495e832c853f882c7250f5c2344c7c868a26233636210b74f714c8e1fd7f72f5"
            },
            "downloads": -1,
            "filename": "flask_http_middleware-0.4.1-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "1f4dd9b893cefe7b2fdfc75c08d8fd5b",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": null,
            "size": 6155,
            "upload_time": "2023-11-03T04:04:59",
            "upload_time_iso_8601": "2023-11-03T04:04:59.433504Z",
            "url": "https://files.pythonhosted.org/packages/84/5a/9222291851feb404a998c2213329dadf2f70f440f3312b89385b9dfd1bfc/flask_http_middleware-0.4.1-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": "",
            "digests": {
                "blake2b_256": "07c64fbeb83acaadd5e67ca2ab504b937a52848f160464263491736d4f599bb3",
                "md5": "10cc27b3fa734badb1bcef52f3934f8c",
                "sha256": "15d5b126b6ebcd4dbe060a339fb13601c9f7c4721b67c9e6b6359ca22eee3d3a"
            },
            "downloads": -1,
            "filename": "flask-http-middleware-0.4.1.tar.gz",
            "has_sig": false,
            "md5_digest": "10cc27b3fa734badb1bcef52f3934f8c",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": null,
            "size": 5871,
            "upload_time": "2023-11-03T04:05:01",
            "upload_time_iso_8601": "2023-11-03T04:05:01.613398Z",
            "url": "https://files.pythonhosted.org/packages/07/c6/4fbeb83acaadd5e67ca2ab504b937a52848f160464263491736d4f599bb3/flask-http-middleware-0.4.1.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2023-11-03 04:05:01",
    "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"
}
danangjoyoo (Agus Danangjoyo)
Elapsed time: 0.13990s