Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use forward_request method in localstack
Best Python code snippet using localstack_python
api.py
Source:api.py 
...37 data=data,38 params=params,39 headers=headers40 )41def forward_request(server, path, params=None, to_manager=False):42 server = db.session.query(Server).filter_by(name=server).first()43 if server is None:44 return jsonify(errors={'name': "Not found"}), 40445 url = server.manager_url if to_manager else server.pdns_url46 headers = {}47 if server.api_key:48 headers['X-API-Key'] = server.api_key49 response = fetch_remote(50 url + path,51 method=request.method,52 data=request.data,53 accept=request.headers.get('Accept'),54 params=params,55 headers=headers56 )57 return forward_remote_response(response)58def eligible_for_passwords():59 return current_user.has_role('admin') or current_user.has_role('edit')60def is_config_password(key):61 """62 Returns true if the config setting identified by key is a password of sorts.63 >>> is_config_password('api-key')64 True65 >>> is_config_password('gmysql-password')66 True67 >>> is_config_password('version-string')68 False69 """70 return key in ('experimental-api-key', 'api-key') or '-password' in key71def remove_passwords_if_ineligible(server):72 obj = dict(server)73 if not eligible_for_passwords():74 del obj['api_key']75 return obj76@mod.route('/servers', methods=['GET'])77@api_auth_required78@roles_required('view')79def server_index():80 ary = [remove_passwords_if_ineligible(obj) for obj in Server.all()]81 return jsonarify(ary)82@mod.route('/servers', methods=['POST'])83@api_auth_required84@roles_required('edit')85def server_create():86 obj = Server()87 obj.mass_assign(request.json)88 if not obj.is_valid:89 return jsonify(errors=obj.validation_errors), 42290 db.session.add(obj)91 db.session.commit()92 return jsonify(**obj.to_dict())93@mod.route('/servers/<string:server>', methods=['GET'])94@api_auth_required95@roles_required('view')96def server_get(server):97 obj = Server.query.filter_by(name=server).first()98 if not obj:99 return jsonify(errors={'name': "Not found"}), 404100 headers = {}101 if obj.api_key:102 headers['X-API-Key'] = obj.api_key103 server = obj.to_dict()104 try:105 s = call_server(obj.name, '/api/v1/servers/localhost', method='GET')106 s.update(server)107 s['id'] = s['_id']108 server = s109 except StandardError:110 pass111 # make sure JS doesn't loop endlessy112 server['version'] = server.get('version', '') # for recursor113 server['local'] = server.get('local', '') # for dnsdist114 server = remove_passwords_if_ineligible(server)115 return jsonify(**server)116@mod.route('/servers/<server>', methods=['DELETE'])117@api_auth_required118@roles_required('edit')119def server_delete(server):120 obj = Server.query.filter_by(name=server).first()121 if not obj:122 return "Not found", 404123 db.session.delete(obj)124 db.session.commit()125 return ""126@mod.route('/servers/<server>', methods=['PUT'])127@api_auth_required128@roles_required('edit')129def server_edit(server):130 obj = Server.query.filter_by(name=server).first()131 if not obj:132 return jsonify(errors={'name': "Not found"}), 404133 obj.mass_assign(request.json)134 if not obj.is_valid:135 return jsonify(errors=obj.validation_errors), 422136 db.session.add(obj)137 db.session.commit()138 return jsonify(**obj.to_dict())139@mod.route('/servers/<server>/zones')140@api_auth_required141@roles_required('view')142def zone_index(server):143 return forward_request(server, '/api/v1/servers/localhost/zones')144@mod.route('/servers/<server>/zones', methods=['POST'])145@api_auth_required146@roles_required('edit')147def zone_create(server):148 return forward_request(server, '/api/v1/servers/localhost/zones')149@mod.route('/servers/<server>/zones/<zone>')150@api_auth_required151@roles_required('view')152def zone_get(server, zone):153 return forward_request(server, '/api/v1/servers/localhost/zones/' + zone)154@mod.route('/servers/<server>/zones/<zone>', methods=['PUT', 'DELETE', 'PATCH'])155@api_auth_required156@roles_required('edit')157def zone_update(server, zone):158 return forward_request(server, '/api/v1/servers/localhost/zones/' + zone)159@mod.route('/servers/<server>/zones/<zone>/axfr-retrieve', methods=['PUT'])160@api_auth_required161@roles_required('edit')162def zone_axfr_retrieve(server, zone):163 return forward_request(server, '/api/v1/servers/localhost/zones/' + zone + '/axfr-retrieve')164@mod.route('/servers/<server>/zones/<zone>/export')165@api_auth_required166@roles_required('view')167def zone_export(server, zone):168 return forward_request(server, '/api/v1/servers/localhost/zones/' + zone + '/export')169@mod.route('/servers/<server>/zones/<zone>/notify', methods=['PUT'])170@api_auth_required171@roles_required('edit')172def zone_notify(server, zone):173 return forward_request(server, '/api/v1/servers/localhost/zones/' + zone + '/notify')174@mod.route('/servers/<server>/search-data')175@api_auth_required176@roles_required('edit')177def server_searchdata(server):178 q = request.values.get('q')179 return forward_request(server, '/api/v1/servers/localhost/search-data', {'q': q})180@mod.route('/servers/<server>/search-log')181@api_auth_required182@roles_required('edit')183def server_loggrep(server):184 q = request.values.get('q')185 return forward_request(server, '/api/v1/servers/localhost/search-log', {'q': q})186@mod.route('/servers/<server>/cache/flush', methods=['PUT'])187@api_auth_required188@roles_required('edit')189def server_flushcache(server):190 domain = request.values.get('domain')191 return forward_request(server, '/api/v1/servers/localhost/cache/flush?domain=' + domain)192@mod.route('/servers/<server>/statistics', methods=['GET'])193@api_auth_required194@roles_required('stats')195def server_stats(server):196 return forward_request(server, '/api/v1/servers/localhost/statistics')197@mod.route('/servers/<server>/config', methods=['GET'])198@api_auth_required199@roles_required('stats')200def server_config(server):201 ary = call_server(server, '/api/v1/servers/localhost/config', method='GET')202 if not eligible_for_passwords():203 ary = [obj for obj in ary if not is_config_password(obj['name'])]204 return jsonarify(ary)205@mod.route('/servers/<server>/config/<config>', methods=['GET'])206@api_auth_required207@roles_required('stats')208def server_config_detail(server, config):209 if is_config_password(config) and not eligible_for_passwords():210 return jsonify(errors={'config': "Forbidden"}), 403211 return forward_request(server, '/api/v1/servers/localhost/config/' + config)212@mod.route('/servers/<server>/config/<config>', methods=['PUT'])213@api_auth_required214@roles_required('edit')215def server_config_edit(server, config):216 return forward_request(server, '/api/v1/servers/localhost/config/' + config)217@mod.route('/servers/<server>/start', methods=['POST'])218@api_auth_required219@roles_required('edit')220def server_start(server):221 return forward_request(server, '/manage/start', params=request.values, to_manager=True)222@mod.route('/servers/<server>/stop', methods=['POST'])223@api_auth_required224@roles_required('edit')225def server_stop(server):226 return forward_request(server, '/manage/stop', params=request.values, to_manager=True)227@mod.route('/servers/<server>/restart', methods=['POST'])228@api_auth_required229@roles_required('edit')230def server_restart(server):231 return forward_request(server, '/manage/restart', params=request.values, to_manager=True)232@mod.route('/servers/<server>/update', methods=['POST'])233@api_auth_required234@roles_required('edit')235def server_update(server):236 return forward_request(server, '/manage/update', params=request.values, to_manager=True)237@mod.route('/servers/<server>/install', methods=['POST'])238@api_auth_required239@roles_required('edit')240def server_install(server):241 return forward_request(server, '/manage/install', params=request.values, to_manager=True)242@mod.route('/me', methods=['GET'])243@api_auth_required244def me_detail():245 data = current_user.to_dict()246 return jsonify(data)247@mod.route('/users', methods=['GET'])248@api_auth_required249@roles_required('view-users')250def users_list():251 ary = User.all()252 return jsonarify(ary)253@mod.route('/users', methods=['POST'])254@api_auth_required255@roles_required('edit-users')...
router.py
Source:router.py
...14 @trade.doc('get trade list')15 def get(self):16 '''List all trades'''1718 return forward_request()192021 @trade.doc('create a trade')22 def post(self):2324 parser: TradeParser = container.get(TradeParser)25 body = parser.parse_create(request)2627 return forward_request(body=body)282930@trade.route('/instructions')31class TradeInstructions(Resource):3233 @trade.doc('get instructions')34 def post(self):3536 parser: TradeParser = container.get(TradeParser)37 body = parser.parse_instructions(request)3839 return forward_request(body=body)404142@trade.route('/<int:trade_id>')43class Trade(Resource):4445 @trade.doc('get trade')46 def get(self, trade_id: int):4748 return forward_request()495051 @trade.doc('delete a trade')52 def delete(self, trade_id: int):5354 return forward_request()555657 @trade.doc('update a trade')58 def put(self, trade_id: int):5960 parser: TradeParser = container.get(TradeParser)61 body = parser.parse_update()6263 return forward_request(body=body)646566@trade.route('/<int:trade_id>/portfolios')67class TradePortfolios(Resource):6869 @trade.doc('add portfolios')70 def post(self, trade_id: int):7172 parser: TradeParser = container.get(TradeParser)73 body = parser.parse_update_portfolios(request)7475 return forward_request(body=body)767778@trade.route('/<int:trade_id>/positions')79class TradePositions(Resource):8081 @trade.doc('get positions')82 def get(self, trade_id: int):8384 parser: TradeParser = container.get(TradeParser)85 params = parser.parse_get_positions(request)8687 return forward_request(params=params)888990 @trade.doc('update positions')91 def put(self, trade_id: int):9293 parser: TradeParser = container.get(TradeParser)94 body = parser.parse_update_positions(request)9596 return forward_request(body=body)979899@trade.route('/<int:trade_id>/prices')100class TradePrices(Resource):101102 @trade.doc('get prices')103 def get(self, trade_id: int):104105 return forward_request()106107108 @trade.doc('update iex prices')109 def post(self, trade_id: int):110111 parser: TradeParser = container.get(TradeParser)112 body = parser.parse_update_positions(request)113114 return forward_request(body=body)115116117@trade.route('/<int:trade_id>/requests')118class TradeRequests(Resource):119120 @trade.doc('get requests')121 def get(self, trade_id: int):122
...
main.py
Source:main.py
1import utils2import time3from flask import Flask, jsonify, request4import os5import requests6import neo4j_queries7import traceback8import project9from project.api import project_blueprint10from artifact.api import artifact_blueprint11from ast.api import ast_blueprint12queue_manager = utils.get_queue_manager()13ast_parsing_queue_manager = utils.get_ast_queue_manager()14dependents_service = utils.get_depends_service()15app = Flask(__name__)16app.register_blueprint(project_blueprint, url_prefix='/project')17app.register_blueprint(artifact_blueprint, url_prefix='/artifact')18app.register_blueprint(ast_blueprint, url_prefix='/ast')19# initiates a dependents search20@app.route('/init/dependents-search/pom', methods=['POST'])21def create_pom_parse_project_task():22 print("here 1")23 print(request)24 print(request.json)25 if not request.json or not 'github_short_url' in request.json:26 return jsonify({'status': 'error'}), 40027 task = {28 'github_short_url': request.json['github_short_url'],29 }30 forward_request = requests.post("http://{}/dependents/project".format(queue_manager), json = {'github_short_url': task.get('github_short_url')})31 if (forward_request.status_code != 200):32 print("Error occurred trying to request parsing of next section")33 return jsonify({"Error from backend service", forward_request.text}), forward_request.status_code34 return jsonify({'task': task}), 20035# initiates a dependents search36@app.route('/init/ast-search/java', methods=['POST'])37def create_ast_parse_project_task():38 # TODO: Don't parse if already parsed or parsing in progress39 # TODO: Should have to post two github urls, the root and the dependent - both should be parsed40 # TODO: Don't parse if project doesn't exist41 42 if not request.json or not 'github_short_url' in request.json or not 'parsing_type' in request.json:43 return jsonify({'Error': 'Must POST JSON request with github_short_url and parsing_type fields'}), 40044 if (request.json['parsing_type'] != 'all' and request.json['parsing_type'] != 'packageclassonly'):45 return jsonify({'Error': "Parsing type must either be 'all' or 'packageclassonly'"}), 40046 task = {47 'github_short_url': request.json['github_short_url'],48 'parsing_type': request.json['parsing_type'],49 }50 forward_request = requests.post("http://{}/java/parse".format(ast_parsing_queue_manager), json = task)51 if (forward_request.status_code != 200):52 print("Error occurred trying to request parsing of next section")53 return jsonify({"Error from backend service", forward_request.text}), forward_request.status_code54 return jsonify({'task': task}), 20055if __name__ == '__main__':...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
