Skip to content

Instantly share code, notes, and snippets.

@Nitrooos
Last active April 28, 2020 21:47
Show Gist options
  • Save Nitrooos/161ccf1ac05d2e0e253a3fe0ff29d3f4 to your computer and use it in GitHub Desktop.
Save Nitrooos/161ccf1ac05d2e0e253a3fe0ff29d3f4 to your computer and use it in GitHub Desktop.
Simple proxy decorator in Python (with Flask and Requests)
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