Flask's Application object has an extensions mapping that can be used to store utilities such as connectors. In our case, we want to store a Session object. We can create a function that will initialize a placeholder in the app.extensions mapping and add a Session object in it:
from requests import Session
def setup_connector(app, name='default', **options):
if not hasattr(app, 'extensions'):
app.extensions = {}
if 'connectors' not in app.extensions:
app.extensions['connectors'] = {}
session = Session()
if 'auth' in options:
session.auth = options['auth']
headers = options.get('headers', {})
if 'Content-Type' not in headers:
headers['Content-Type'] = 'application/json'
session.headers.update(headers)
app.extensions['connectors'][name] = session
return session
def get_connector(app, name='default'):
return app.extensions['connectors'][name]
In this example, the setup_connector() function will create a Session object and store it in the app's extensions mapping. The created Session will set the Content-Type header to application/json by default, so it's suitable for sending data to JSON-based microservices.
Using the session from a view can then be done with the get_connector() function once it has been set up on the app. In the following example, a Flask app running on port 5001 will synchronously call a microservice running on 5000 to serve its content:
from flask import Flask, jsonify
app = Flask(__name__)
setup_connector(app)
@app.route('/api', methods=['GET', 'POST'])
def my_microservice():
with get_connector(app) as conn:
sub_result = conn.get('http://localhost:5000/api').json()
return jsonify({'result': sub_result, 'Hello': 'World!'})
if __name__ == '__main__':
app.run(port=5001)
A call to the service will propagate a call to the other service:
$ curl http://127.0.0.1:5001/api
{
"Hello": "World!",
"result": {
"Hello": "World!",
"result": "OK"
}
}
This naive implementation is based on the hypothesis that everything will go smoothly. But what will happen if the microservice that's called lags and takes 30 seconds to return?
By default, requests will hang indefinitely until the answer is ready, which is not a behavior we'd want when calling microservices. The timeout option is useful in this case. Used when making a request, it will raise a ReadTimeout in case the remote server fails to answer promptly.
In the following example, we drop the call if it's hanging for more than 2 seconds:
from requests.exceptions import ReadTimeout
@app.route('/api', methods=['GET', 'POST'])
def my_microservice():
with get_connector(app) as conn:
try:
result = conn.get('http://localhost:5000/api',timeout=2.0).json()
except ReadTimeout:
result = {}
return jsonify({'result': result, 'Hello': 'World!'})
Of course, what should be done when a timeout happens depends on your service logic. In this example, we silently ignore the problem and send back an empty result. But maybe in other cases, you will need to raise an error. In any case, handling timeouts is mandatory if you want to build a robust service-to-service link.
The other error that can happen is when the connection completely drops, or the remote server is not reachable at all. Requests will retry several times and eventually will raise a ConnectionError you need to catch:
from requests.exceptions import ReadTimeout, ConnectionError
@app.route('/api', methods=['GET', 'POST'])
def my_microservice():
with get_connector(app) as conn:
try:
result = conn.get('http://localhost:5000/api',
timeout=2.).json()
except (ReadTimeout, ConnectionError):
result = {}
return jsonify({'result': result, 'Hello': 'World!'})
Since it's good practice to always use the timeout option, a better way would be to set a default one at the session level, so we don't have to provide it on every request call.
To do this, the requests library has a way to set up custom transport adapters, where you can define a behavior for a given host the session will call. It can be used to create a general timeout, but also to offer a retries option in case you want to tweak how many retries should be done when the service is not responsive.
Back to our setup_connector() function. Using an adapter, we can add timeout and retries options that will be used by default for all requests:
from requests.adapters import HTTPAdapter
class HTTPTimeoutAdapter(HTTPAdapter):
def __init__(self, *args, **kw):
self.timeout = kw.pop('timeout', 30.)
super().__init__(*args, **kw)
def send(self, request, **kw):
timeout = kw.get('timeout')
if timeout is None:
kw['timeout'] = self.timeout
return super().send(request, **kw)
def setup_connector(app, name='default', **options):
if not hasattr(app, 'extensions'):
app.extensions = {}
if 'connectors' not in app.extensions:
app.extensions['connectors'] = {}
session = Session()
if 'auth' in options:
session.auth = options['auth']
headers = options.get('headers', {})
if 'Content-Type' not in headers:
headers['Content-Type'] = 'application/json'
session.headers.update(headers)
retries = options.get('retries', 3)
timeout = options.get('timeout', 30)
adapter = HTTPTimeoutAdapter(max_retries=retries, timeout=timeout)
session.mount('http://', adapter)
app.extensions['connectors'][name] = session
return session
The session.mount(host, adapter) call will tell requests to use the HTTPTimeoutAdapter every time a request for any HTTP service is made. The http:// value for the host is a catch-all in this case.
The beautiful thing about the mount() function is that the session behavior can be tweaked on a service-per-service basis depending on your app logic. For instance, you can mount another instance on the adapter for a particular host if you need to set up some custom timeouts and retries values:
adapter2 = HTTPTimeoutAdapter(max_retries=1, timeout=1.)
session.mount('http://myspecial.service', adapter2)
Thanks to this pattern, a single request ;Session object can be instantiated into your application to interact with many other HTTP services.