Routing Requests With Flask, With Extra Authentication Headers
Flask
Gunicorn
Authentication
HTTP methods
Response chunk generation
This article provides a step-by-step guide on how to set up a Flask application with Gunicorn, implement extra authentication headers support, enable both GET and POST HTTP methods, and create a response chunk generation function.
Run the following like:
pip3 install gunicorn
gunicorn --bind localhost:8001 <file_name_without_extension>:app
import argparse
import requests
from flask import Flask, Response, request
import json
= argparse.ArgumentParser(description='Argument Parser with Default Parameters')
parser '--source_port', type=int, default=8000, help='Source Port Number')
parser.add_argument(= parser.parse_known_args()
args, _ print("Source Port:", args.source_port)
= args.source_port
source_port = Flask(__name__)
app = requests.Session()
sess = "auth_token"
AUTH_TOKEN = "Auth"
AUTH_HEADER_KEY = "GET"
GET = "POST"
POST = [GET, POST]
ALLOWED_METHODS @app.route('/', defaults={'path': ''}, methods=ALLOWED_METHODS)
@app.route('/<path:path>', methods=ALLOWED_METHODS)
def chat_completions(path):
= f'http://localhost:{source_port}{request.full_path}'
url # full path is prefixed with /
= dict(request.headers)
request_headers = request_headers.get(AUTH_HEADER_KEY, None)
auth if auth != AUTH_TOKEN:
return Response(json.dumps({"state": "unauthorized", "message": "Unauthorized access"}), status = 401)
= {k:v for k,v in request_headers.items() if k !=AUTH_HEADER_KEY}
no_auth_headers if request.method == GET:
= sess.get(url, stream=True, headers=no_auth_headers)
response else:
= sess.post(url, stream=True, headers=no_auth_headers, data=request.data, files=request.files, json=request.json)
response # form is not accepted.
# response = sess.post(url, stream=True, headers=no_auth_headers, data=request.data, form=request.form, files=request.files, json=request.json)
def generate():
for chunk in response.iter_content(chunk_size=1024):
yield chunk
return Response(generate(), content_type=response.headers['content-type'], headers = dict(response.headers))
if __name__ == '__main__':
app.run()