How to use query_params method in localstack
Best Python code snippet using localstack_python
api_service.py
Source:api_service.py 
1# DO NOT EDIT THIS FILE. This file will be overwritten when re-running go-raml.2from .Capacity import Capacity3from .Farmer import Farmer4from .unhandled_api_error import UnhandledAPIError5from .unmarshall_error import UnmarshallError6class ApiService:7 def __init__(self, client):8 self.client = client9 def RegisterFarmer(self, headers=None, query_params=None, content_type="application/json"):10 """11 Register a farmer12 It is method for GET /api/farmer_create13 """14 if query_params is None:15 query_params = {}16 uri = self.client.base_url + "/api/farmer_create"17 return self.client.get(uri, None, headers, query_params, content_type)18 def UpdateFarmer(self, headers=None, query_params=None, content_type="application/json"):19 """20 Update a farmer21 It is method for GET /api/farmer_update22 """23 if query_params is None:24 query_params = {}25 uri = self.client.base_url + "/api/farmer_update"26 return self.client.get(uri, None, headers, query_params, content_type)27 def GetFarmer(self, iyo_organization, headers=None, query_params=None, content_type="application/json"):28 """29 Get detail about a farmer30 It is method for GET /api/farmers/{iyo_organization}31 """32 if query_params is None:33 query_params = {}34 uri = self.client.base_url + "/api/farmers/" + iyo_organization35 resp = self.client.get(uri, None, headers, query_params, content_type)36 try:37 if resp.status_code == 200:38 return Farmer(resp.json()), resp39 message = "unknown status code={}".format(resp.status_code)40 raise UnhandledAPIError(response=resp, code=resp.status_code, message=message)41 except ValueError as msg:42 raise UnmarshallError(resp, msg)43 except UnhandledAPIError as uae:44 raise uae45 except Exception as e:46 raise UnmarshallError(resp, e.message)47 def ListFarmers(self, headers=None, query_params=None, content_type="application/json"):48 """49 List Farmers50 It is method for GET /api/farmers51 """52 if query_params is None:53 query_params = {}54 uri = self.client.base_url + "/api/farmers"55 resp = self.client.get(uri, None, headers, query_params, content_type)56 try:57 if resp.status_code == 200:58 resps = []59 for elem in resp.json():60 resps.append(Farmer(elem))61 return resps, resp62 message = "unknown status code={}".format(resp.status_code)63 raise UnhandledAPIError(response=resp, code=resp.status_code, message=message)64 except ValueError as msg:65 raise UnmarshallError(resp, msg)66 except UnhandledAPIError as uae:67 raise uae68 except Exception as e:69 raise UnmarshallError(resp, e.message)70 def UpdateActualUsedCapacity(self, data, node_id, headers=None, query_params=None, content_type="application/json"):71 """72 Set the actual usage of the capacity of a node73 It is method for PUT /api/nodes/{node_id}/actual74 """75 if query_params is None:76 query_params = {}77 uri = self.client.base_url + "/api/nodes/" + node_id + "/actual"78 return self.client.put(uri, data, headers, query_params, content_type)79 def UpdateReservedCapacity(self, data, node_id, headers=None, query_params=None, content_type="application/json"):80 """81 Mark some capacity on a node to be reserved82 It is method for PUT /api/nodes/{node_id}/reserved83 """84 if query_params is None:85 query_params = {}86 uri = self.client.base_url + "/api/nodes/" + node_id + "/reserved"87 return self.client.put(uri, data, headers, query_params, content_type)88 def GetCapacity(self, node_id, headers=None, query_params=None, content_type="application/json"):89 """90 Get detail about capacity of a node91 It is method for GET /api/nodes/{node_id}92 """93 if query_params is None:94 query_params = {}95 uri = self.client.base_url + "/api/nodes/" + node_id96 resp = self.client.get(uri, None, headers, query_params, content_type)97 try:98 if resp.status_code == 200:99 return Capacity(resp.json()), resp100 message = "unknown status code={}".format(resp.status_code)101 raise UnhandledAPIError(response=resp, code=resp.status_code, message=message)102 except ValueError as msg:103 raise UnmarshallError(resp, msg)104 except UnhandledAPIError as uae:105 raise uae106 except Exception as e:107 raise UnmarshallError(resp, e.message)108 def ListCapacity(self, headers=None, query_params=None, content_type="application/json"):109 """110 List all the nodes capacity111 It is method for GET /api/nodes112 """113 output = []114 for node in self.ListCapacityGenerator(headers, query_params, content_type):115 output.append(Capacity(node))116 return output117 def ListCapacityGenerator(self, headers=None, query_params=None, content_type="application/json", size=10):118 """119 Yields nodes120 """121 if query_params is None:122 query_params = {}123 def get_page(page):124 query_params.update({"page": page, "per_page": size})125 uri = self.client.base_url + "/api/nodes"126 resp = self.client.get(uri, None, headers, query_params, content_type)127 try:128 if resp.status_code == 200:129 yield resp130 for elem in resp.json():131 yield elem132 else:133 message = "unknown status code={}".format(resp.status_code)134 raise UnhandledAPIError(response=resp, code=resp.status_code, message=message)135 except ValueError as msg:136 raise UnmarshallError(resp, msg)137 except UnhandledAPIError as uae:138 raise uae139 except Exception as e:140 raise UnmarshallError(resp, e.message)141 nodes = get_page(page=1)142 resp = next(nodes)143 pages = resp.headers.get("page", 1)144 for node in nodes:145 yield node146 for i in range(2, int(pages) + 1):147 nodes = get_page(page=i)148 next(nodes)149 for node in nodes:150 yield node151 def RegisterCapacity(self, data, headers=None, query_params=None, content_type="application/json"):152 """153 Register a node capacity154 It is method for POST /api/nodes155 """156 if query_params is None:157 query_params = {}158 uri = self.client.base_url + "/api/nodes"159 resp = self.client.post(uri, data, headers, query_params, content_type)160 try:161 if resp.status_code == 201:162 return Capacity(resp.json()), resp163 message = "unknown status code={}".format(resp.status_code)164 raise UnhandledAPIError(response=resp, code=resp.status_code, message=message)165 except ValueError as msg:166 raise UnmarshallError(resp, msg)167 except UnhandledAPIError as uae:168 raise uae169 except Exception as e:...
api.py
Source:api.py
1import flask2from flask import request, jsonify3import accounts4import quotes5import buy6import sell7import setbuy8import setsell9import logs10app = flask.Flask(__name__)11app.config["DEBUG"] = True12#Database queries13@app.route('/useraccount', methods=['POST'])14def get_user():15 query_params = request.get_json()16 user_id = query_params["id"]17 result = accounts.find_user_account(user_id)18 return jsonify(result)19@app.route('/userstocks', methods=['POST'])20def get_user_stocks():21 query_params = request.get_json()22 user_id = query_params["id"]23 24 result = accounts.get_user_stocks(user_id)25 # print("The info is --> ", result)26 return jsonify(result)27#Generic functions28@app.route('/ADD', methods=['POST'])29def add_funds_to_account():30 query_params = request.get_json()31 user_id = query_params["id"]32 amount = float(query_params["amount"])33 #update user account with added money34 result = accounts.add_funds(user_id, amount)35 response = {36 "userid": result["id"],37 "balance": result["amount"]38 }39 return jsonify(response)40@app.route('/QUOTE', methods=["POST"])41def quote_stock():42 query_params = request.get_json()43 print(query_params)44 user_id = query_params["id"]45 stock_name = query_params["stock"]46 47 result = quotes.get_quote(user_id, stock_name)48 print("The stock quote result is --> ", result)49 return jsonify(result)50#Buy51@app.route('/BUY', methods=['POST'])52def buy_stock():53 query_params = request.get_json()54 user_id = query_params["id"]55 stock_name = query_params["stock"]56 amount = float(query_params["amount"])57 print("Buy for ", user_id)58 result = buy.start_buy(user_id, stock_name, amount)59 print(jsonify(result))60 return jsonify(result)61@app.route('/COMMIT_BUY', methods=['POST'])62def commit_buy_stock():63 query_params = request.get_json()64 user_id = query_params["id"]65 print("Commiting most recent buy for ", user_id)66 result = buy.commit_buy(user_id)67 print(jsonify(result))68 return jsonify(result)69@app.route('/CANCEL_BUY', methods=['POST'])70def cancel_buy_stock():71 query_params = request.get_json()72 user_id = query_params["id"]73 print("cancelling most recent buy for ", user_id)74 result = buy.cancel_buy(user_id)75 return jsonify(result)76#Sell77@app.route('/SELL', methods=['POST'])78def sell_stock():79 query_params = request.get_json()80 user_id = query_params["id"]81 stock_name = query_params["stock"]82 amount = float(query_params["amount"])83 print("Executing sell for ", user_id, stock_name)84 result = sell.start_sell(user_id, stock_name, amount)85 return jsonify(result)86@app.route('/COMMIT_SELL', methods=['POST'])87def commit_sell_stock():88 query_params = request.get_json()89 user_id = query_params["id"]90 91 result = sell.commit_sell(user_id)92 return jsonify(result)93@app.route('/CANCEL_SELL', methods=['POST'])94def cancel_sell_stock():95 query_params = request.get_json()96 user_id = query_params["id"]97 print("Cancel recent sell for ", user_id)98 result = sell.cancel_sell(user_id)99 return jsonify(result)100#Set Buy101@app.route('/SET_BUY_AMOUNT', methods=['POST'])102def set_buy_stock():103 query_params = request.get_json()104 user_id = query_params["id"]105 stock_name = query_params["stock"]106 amount = float(query_params["amount"])107 #query for user cash amount108 #if sufficient:109 # create a log of trans110 # create reserve account holding allocated funds?111 # update db with decreased cash amount, trigger amount112 result = setbuy.set_buy_amount(user_id, stock_name, amount)113 return jsonify(result)114@app.route('/SET_BUY_TRIGGER', methods=['POST'])115def set_buy_stock_trigger():116 query_params = request.get_json()117 user_id = query_params["id"]118 stock_name = query_params["stock"]119 amount = float(query_params["amount"])120 #query for quoteserver stock price121 #create and commit a buy action if the stock price <= trigger amount122 result = setbuy.set_buy_trigger(user_id, stock_name, amount)123 return jsonify(result)124@app.route('/CANCEL_SET_BUY', methods=['POST'])125def cancel_set_buy_stock():126 query_params = request.get_json()127 user_id = query_params["id"]128 stock_name = query_params["stock"]129 #query transactions (or user account?) for a set buy amount130 #if exists,131 # account is reset to the value before the set buy132 # remove the buy trigger133 # create a transaction log134 result = setbuy.cancel_set_buy(user_id, stock_name)135 return jsonify(result)136#Set Sell137@app.route('/SET_SELL_AMOUNT', methods=['POST'])138def set_sell_stock_amount():139 query_params = request.get_json()140 user_id = query_params["id"]141 stock_name = query_params["stock"]142 amount = float(query_params["amount"])143 #query for user stock amount144 #if sufficient:145 # create a log of trans146 # create reserve account holding allocated stocks?147 # update db with decreased stock amount, trigger amount148 result = setsell.set_sell_amount(user_id, stock_name, amount)149 return jsonify(result)150@app.route('/SET_SELL_TRIGGER', methods=['POST'])151def set_sell_stock_trigger():152 query_params = request.get_json()153 user_id = query_params["id"]154 stock_name = query_params["stock"]155 amount = float(query_params["amount"])156 #query for quoteserver stock price157 #create and commit a sell action if the stock price >= trigger amount158 result = setsell.set_sell_trigger(user_id, stock_name, amount)159 return jsonify(result)160@app.route('/CANCEL_SET_SELL', methods=['POST'])161def cancel_set_sell_stock():162 query_params = request.get_json()163 user_id = query_params["id"]164 stock_name = query_params["stock"]165 #query transactions (or user account?) for a set sell amount166 #if exists,167 # account is reset to the value before the set sell168 # remove the sell trigger169 # create a transaction log170 result = setsell.cancel_set_sell(user_id, stock_name)171 return jsonify(result)172#Logs173@app.route('/USER_DUMPLOG', methods=['POST'])174def user_dumplog_endpoint():175 query_params = request.get_json()176 user_id = query_params['id']177 output_file = query_params.get('filename')178 result = logs.user_dumplog(user_id, output_file)179 return result180@app.route('/DUMPLOG', methods=['POST'])181def dumplog_endpoint():182 query_params = request.get_json()183 output_file = query_params.get('filename')184 result = logs.dumplog(output_file)185 return result186@app.route('/DISPLAY_SUMMARY', methods=['POST'])187def display_summary_endpoint():188 query_params = request.get_json()189 user_id = query_params["id"]190 result = logs.display_summary(user_id)191 return jsonify(result)...
wiki_req.py
Source:wiki_req.py
1import requests2from datetime import datetime3import wiki_analysis as wa4url = "https://simple.wikipedia.org/w/api.php"5user_agent = "iui-wikipedia (https://github.com/yelouis/IUI_Wiki_Comp)"6query_base = query_params = {7 "action": "query",8 "format": "json",9}10headers = {"User-Agent": user_agent}11rv_limit = 1012def get_article_properties(pageid):13 # Returns the article's current revision ID, parent revision ID, and the past rv_limit revision IDs14 query_params = query_base.copy()15 query_params["pageids"] = pageid16 query_params["rvprop"] = "ids"17 query_params["rvlimit"] = rv_limit18 query_params["prop"] = "revisions"19 request = requests.get(url=url, params=query_params, headers=headers).json()20 revisions = request["query"]["pages"][str(pageid)]["revisions"]21 revision_ids = [cell["revid"] for cell in revisions]22 cid = request["query"]["pages"][str(pageid)]["revisions"][0]["revid"]23 pid = request["query"]["pages"][str(pageid)]["revisions"][0]["parentid"]24 return revision_ids, cid, pid25def get_creation_date(pageid):26 # Returns a timestamp of the article's creation27 query_params = query_base.copy()28 query_params["pageids"] = pageid29 query_params["rvprop"] = "timestamp"30 query_params["rvlimit"] = 131 query_params["rvdir"] = "newer"32 query_params["prop"] = "revisions"33 request = requests.get(url=url, params=query_params, headers=headers).json()34 revisions = request["query"]["pages"][str(pageid)]["revisions"]35 timestamp = datetime.strptime(revisions[0]["timestamp"], "%Y-%m-%dT%H:%M:%SZ")36 return timestamp37def get_revision_date(revision_id, pageid):38 # Returns the timestamp of a given revision39 query_params = query_base.copy()40 query_params["revids"] = revision_id41 query_params["rvprop"] = "timestamp|ids"42 query_params["prop"] = "revisions"43 request = requests.get(url=url, params=query_params, headers=headers).json()44 revisions = request["query"]["pages"][str(pageid)]["revisions"]45 timestamp = datetime.strptime(revisions[0]["timestamp"], "%Y-%m-%dT%H:%M:%SZ")46 return timestamp47def get_text(pageid):48 # Returns plain text of the article's body49 query_params = query_base.copy()50 query_params["pageids"] = pageid51 query_params["prop"] = "extracts"52 request = requests.get(url=url, params=query_params, headers=headers).json()53 text = request["query"]["pages"][str(pageid)]["extract"]54 return text55def get_images(pageid):56 # Returns a list of images in the article57 query_params = query_base.copy()58 query_params["pageids"] = pageid59 query_params["prop"] = "images"60 request = requests.get(url=url, params=query_params, headers=headers).json()61 images = request["query"]["pages"][str(pageid)]["images"]62 return images63def get_external_links(pageid):64 # Returns list of external links within the article65 query_params = query_base.copy()66 query_params["pageids"] = pageid67 query_params["prop"] = "extlinks"68 request = requests.get(url=url, params=query_params, headers=headers).json()69 extlinks = request["query"]["pages"][str(pageid)]["extlinks"]70 return extlinks71def get_internal_links(pageid):72 # Returns list of internal links within the article73 query_params = query_base.copy()74 query_params["pageids"] = pageid75 query_params["prop"] = "links"76 request = requests.get(url=url, params=query_params, headers=headers).json()77 links = request["query"]["pages"][str(pageid)]["links"]78 return links79def get_contributors(pageid, admin=False):80 # Returns all contributors to the article. If admin is true then returns only the trusted editors.81 query_params = query_base.copy()82 query_params["pageids"] = pageid83 query_params["prop"] = "contributors"84 if admin:85 query_params["pcgroup"] = "bureaucrat"86 request = requests.get(url=url, params=query_params, headers=headers).json()87 contribs = request["query"]["pages"][str(pageid)]["contributors"]88 return contribs89def main():90 tester = wa.WikipediaArticle(17338)91if __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?
