try: r = http_request(http_client, 'GET', url) except: flash('Failed to fetch RSS feed', 'error') return redirect(url_for('import_rss_form')) try: rss = RSSParser.parse(r.text) except: flash('Invalid RSS feed', 'error') return redirect(url_for('import_rss_form')) for item in rss.channel.items: r = http_request(http_client, 'POST', API_SERVER + '/api/posts', {'user_id': session['user_id'], 'title': str(item.title.content), 'body': str(item.description.content)}) result = r.json() if (error_message := result.get('error', None)) != None: flash(error_message, 'error') return redirect(url_for('import_rss_form'))
return redirect(url_for('posts_list'))
api-key的传递方式就是挂载,只要是API_SERVER开头的就会自动发送X-API-KEY
1 2 3 4 5 6 7 8
classApiAdapter(HTTPAdapter): defadd_headers(self, request, **kwargs): request.headers['X-API-KEY'] = API_KEY definit_http_client(): s = requests.Session() s.mount(API_SERVER, ApiAdapter()) return s
查看源码也可以发现:
1 2 3 4 5 6 7 8 9 10 11 12 13 14
classSession(SessionRedirectMixin): ... # Get the appropriate adapter to use adapter = self.get_adapter(url=request.url)
# redirect to get_adapter defget_adapter(self, url): """ Returns the appropriate connection adapter for the given URL. :rtype: requests.adapters.BaseAdapter """ for prefix, adapter inself.adapters.items(): if url.lower().startswith(prefix.lower()): return adapter