Last active
April 28, 2020 21:47
-
-
Save Nitrooos/161ccf1ac05d2e0e253a3fe0ff29d3f4 to your computer and use it in GitHub Desktop.
Simple proxy decorator in Python (with Flask and Requests)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests | |
from flask import Flask, request, Response, make_response | |
app = Flask(__name__) | |
DESTINATION_URL = 'http://any-external-service/' | |
def proxy_1(view_function): | |
def wrapper(path): | |
request_method = getattr(requests, request.method.lower()) | |
return request_method(f'{DESTINATION_URL}/{path}').content | |
return wrapper | |
def proxy_2(view_function): | |
def wrapper(path): | |
request_method = getattr(requests, request.method.lower()) | |
response = request_method(f'{DESTINATION_URL}/{path}', | |
data=request.form or request.json or request.data, | |
headers=request.headers, | |
) | |
return response.content, response.status_code, response.headers.items() | |
return wrapper | |
def proxy_2_1(view_function): | |
def wrapper(path): | |
request_method = getattr(requests, request.method.lower()) | |
response = clean_hop_by_hop_headers( | |
request_method(f'{DESTINATION_URL}/{path}', | |
data=request.form or request.json or request.data, | |
headers=request.headers, | |
) | |
) | |
return response.content, response.status_code, response.headers.items() | |
return wrapper | |
def proxy_3(view_function): | |
def wrapper(path): | |
request_method = getattr(requests, request.method.lower()) | |
raw_response = clean_hop_by_hop_headers( | |
request_method(f'{DESTINATION_URL}/{path}', | |
data=request.form or request.json or request.data, | |
headers=request.headers, | |
), | |
) | |
processed_response = view_function(raw_response) | |
if isinstance(processed_response, Response): | |
return processed_response | |
return requests_to_flask_response(raw_response) | |
return wrapper | |
def proxy_4(proxy_path=''): | |
def real_proxy(view_function): | |
def wrapper(path): | |
final_path = proxy_path if proxy_path != '' else path | |
request_method = getattr(requests, request.method.lower()) | |
raw_response = clean_hop_by_hop_headers( | |
request_method(f'{DESTINATION_URL}/{final_path}', | |
data=request.form or request.json or request.data, | |
headers=request.headers, | |
), | |
) | |
processed_response = view_function(raw_response) | |
if isinstance(processed_response, Response): | |
return processed_response | |
return requests_to_flask_response(raw_response) | |
return wrapper | |
return real_proxy | |
def clean_hop_by_hop_headers(response): | |
hop_by_hop_headers = [ | |
'Connection', 'Keep-Alive', 'TE', 'Trailer', 'Transfer-Encoding', 'Upgrade', | |
'Proxy-Authorization', 'Proxy-Authenticate' | |
] | |
for header in hop_by_hop_headers: | |
if header in response.headers: | |
del response.headers[header] | |
return response | |
def requests_to_flask_response(response): | |
return response.content, response.status_code, response.headers.items() | |
@app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE']) | |
@proxy_4(proxy_path='/api') | |
def test_view(proxied_response): | |
response = make_response({ | |
'status_message': proxied_response.json()['status'] | |
}) | |
response.status_code = 200 | |
response.headers = { **proxied_response.headers, 'x-foo': 'bar' }.items() | |
return response |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment