From 4b23261bd1f7fd2daa6fcb8cd1829aa625804243 Mon Sep 17 00:00:00 2001 From: Linus E <75929322+Rripped@users.noreply.github.com> Date: Mon, 28 Mar 2022 18:55:52 +0200 Subject: [PATCH] api_handler functions completed (tbd: testing) --- telegram_bot/api_handler.py | 189 +++++++++++++++++++++++++++++++++--- 1 file changed, 174 insertions(+), 15 deletions(-) diff --git a/telegram_bot/api_handler.py b/telegram_bot/api_handler.py index aa9b4da..45c5de4 100644 --- a/telegram_bot/api_handler.py +++ b/telegram_bot/api_handler.py @@ -15,11 +15,31 @@ import requests as r class API_Handler: #class for communicating with webservice to get data from database - def __init__(self, db_adress): - self.db_adress = db_adress - return + def __init__(self, db_adress, email, password): + """initializes the API_Handler class - def get_auth_token(self, email, password): + Args: + db_adress (string): adress of the database + email (string): email of the user + password (string): password of the user + """ + self.db_adress = db_adress + + payload = {'email': email, 'password': password} + with r.Session() as s: + p = s.post(self.db_adress + "/user/login", json=payload) + if p.status_code == 200: + self.token = p.json()["data"]['token'] + else: + self.token = None + + def reauthorize(self, email, password): + """reauthorizes the user + + Args: + email (string): email of the user + password (string): password of the user + """ payload = {'email': email, 'password': password} with r.Session() as s: p = s.post(self.db_adress + "/user/login", json=payload) @@ -27,28 +47,167 @@ class API_Handler: self.token = p.json()["data"]['token'] return p.json()["data"]['token'] else: - #print server response - print(p.text) + self.token = None return None - def get_user_keywords(self, token, user_id): + def get_user_keywords(self, user_id): + """gets the keywords of the user + + Args: + user_id (int): id of the user + + Returns: + list: list of keywords + """ with r.Session() as s: - headers = {'Authorization': 'Bearer ' + token + ":" + user_id} - p = s.get(self.db_adress + "/api/keywords", headers=headers) - keywords_json = p.json()["data"] + headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} + req = s.get(self.db_adress + "/keywords", headers=headers) + print(req.status_code) + keywords_json = req.json()["data"] keywords = [] for keyword in keywords_json: keywords.append(keyword["keyword"]) + return keywords + def set_keyword(self, user_id, keyword): + """sets the keyword of the user + + Args: + user_id (int): id of the user + keyword (int): keyword of the user + + Returns: + int: status code + """ + with r.Session() as s: + headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} + req = s.post(self.db_adress + "/keyword", json={"keyword": keyword}, headers=headers) + + return req.status_code + + def delete_keyword(self, user_id, keyword): + """deletes the keyword of the user + + Args: + user_id (int): id of the user + keyword (string): keyword of the user + + Returns: + int: status code + """ + with r.Session() as s: + headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} + req = s.delete(self.db_adress + "/keyword", json={"keyword": keyword}, headers=headers) + + return req.status_code + + def get_user_shares(self, user_id): + """gets the shares of the user + + Args: + user_id (int): id of the user + + Returns: + list: list of shares + """ + with r.Session() as s: + headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} + req = s.get(self.db_adress + "/shares", headers=headers) + shares_json = req.json()["data"] + shares = [] + for share in shares_json: + shares.append(share["symbol"]) + + return shares + + def set_share(self, user_id, symbol): + """sets the share of the user + + Args: + user_id (int): id of the user + symbol (string): symbol of the share + + Returns: + int: status code + """ + with r.Session() as s: + headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} + req = s.post(self.db_adress + "/share", json={"symbol": symbol}, headers=headers) + return req.status_code + + def delete_share(self, user_id, symbol): + """deletes the share of the user + + Args: + user_id (int): id of the user + symbol (string): symbol of the share + + Returns: + int: status code + """ + with r.Session() as s: + headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} + req = s.delete(self.db_adress + "/share", json={"symbol": symbol}, headers=headers) + return req.status_code + + def get_user_transactions(self, user_id): + """gets the transactions of the user + + Args: + user_id (int): id of the user + + Returns: + dict: dictionary of transactions + """ + with r.Session() as s: + headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} + req = s.get(self.db_adress + "/transactions", headers=headers) + transactions_dict = dict(req.json()["data"]) + return transactions_dict + + def set_transaction(self, user_id, count, price, symbol, timestamp): + """sets the transaction of the user + + Args: + user_id (int): id of the user + count (int): count of the transaction + price (float): price of the transaction + symbol (string): symbol of the transaction + timestamp (string): timestamp of the transaction + + Returns: + int: status code + """ + with r.Session() as s: + headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} + transaction = {"count": count, "price": price, "symbol": symbol, "time": timestamp} + req = s.post(self.db_adress + "/transaction", json=transaction, headers=headers) + return req.status_code + + def get_user_portfolio(self, user_id): + """gets the portfolio of the user + + Args: + user_id (int): id of the user + + Returns: + dict: dictionary of portfolio + """ + with r.Session() as s: + headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} + req = s.get(self.db_adress + "/portfolio", headers=headers) + portfolio_dict = dict(req.json()["data"]) + return portfolio_dict + + + if __name__ == "__main__": print("This is a module for the telegram bot. It is not intended to be run directly.") - handler = API_Handler("https://aktienbot.flokaiser.com/api") - try: - handler.get_auth_token("bot@example.com", "bot") - except None: - print("Could not connect to server.") + handler = API_Handler("https://aktienbot.flokaiser.com/api", "bot@example.com", "bot") + keywords = handler.get_user_keywords(user_id = 1709356058) #user_id is mine (Linus) + print(keywords) sys.exit(1) \ No newline at end of file