""" script for communicating with webservice to get data from database """ __author__ = "Florian Kellermann, Linus Eickhoff" __date__ = "16.03.2022" __version__ = "0.0.1" __license__ = "None" #side-dependencies: none #Work in Progress import sys import os import requests as r class API_Handler: """class for interacting with the api webservice Attributes: db_adress (string): adress of the database token (string): auth token for api Methods: reauthorize(email, password): set new credentials get_user_keywords(user_id): gets the keywords of the user set_keyword(user_id, keyword): sets the keyword of the user delete_keyword(user_id, keyword): deletes the keyword of the user get_user_shares(user_id): gets the shares of the user set_share(user_id, symbol): sets the share of the user delete_share(user_id, symbol): deletes the share of the user get_user_transactions(user_id): gets the transactions of the user set_transaction(user_id, transaction): sets the transaction of the user delete_transaction(user_id, transaction): deletes the transaction of the user """ def __init__(self, db_adress, email, password): """initializes the API_Handler class 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: print("Error: " + str(p.status_code) + " invalid credentials") self.token = None def reauthorize(self, email, password): """set new credentials 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) if p.status_code == 200: self.token = p.json()["data"]['token'] return p.json()["data"]['token'] else: self.token = None return None def get_user(self, user_id, max_retries=10): """gets the shares of the user Args: user_id (int): id of the user max_retries (int): max retries for the request Returns: json: json of user infos """ if max_retries <= 0: return None with r.Session() as s: headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} req = s.get(self.db_adress + "/user", headers=headers) if(req.status_code == 200): return req.json()["data"] else: return self.get_user(user_id, max_retries-1) def get_all_users(self, max_retries=10): """gets all users Args: max_retries (int): max retries for the request Returns: list: list of users """ if max_retries <= 0: return None with r.Session() as s: headers = {'Authorization': 'Bearer ' + self.token} req = s.get(self.db_adress + "/users", headers=headers) if(req.status_code == 200): return req.json()["data"] else: return self.get_all_users(max_retries-1) def get_user_keywords(self, user_id, max_retries=10): """gets the keywords of the user Args: user_id (int): id of the user max_retries (int): max retries for the request Returns: list: list of keywords """ if max_retries <= 0: return None keywords = [] with r.Session() as s: headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} req = s.get(self.db_adress + "/keywords", headers=headers) if(req.status_code == 200): keywords_json = req.json()["data"] for keyword in keywords_json: keywords.append(keyword["keyword"]) return keywords else: return self.get_user_keywords(user_id, max_retries-1) 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, max_retries=10): """gets the shares of the user Args: user_id (int): id of the user max_retries (int): max retries for the request Returns: list: list of shares """ if max_retries <= 0: return None with r.Session() as s: headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} req = s.get(self.db_adress + "/shares", headers=headers) if(req.status_code == 200): shares_json = req.json()["data"] shares = [] for share in shares_json: shares.append(share["symbol"]) return shares else: return self.get_user_shares(user_id, max_retries-1) 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, max_retries=10): """gets the transactions of the user Args: user_id (int): id of the user max_retries (int): max retries for the request Returns: dict: dictionary of transactions """ if max_retries <= 0: return None with r.Session() as s: headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} req = s.get(self.db_adress + "/transactions", headers=headers) if req.status_code == 200: transactions_dict = dict(req.json()["data"]) return transactions_dict else: return self.get_user_transactions(user_id, max_retries-1) 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, max_retries=10): """gets the portfolio of the user Args: user_id (int): id of the user max_retries (int): max retries for the request Returns: dict: dictionary of portfolio """ if max_retries <= 0: return None with r.Session() as s: headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} req = s.get(self.db_adress + "/portfolio", headers=headers) if req.status_code == 200: portfolio_dict = dict(req.json()["data"]) return portfolio_dict else: return self.get_user_portfolio(user_id, max_retries-1) def set_cron_interval(self, user_id, cron_interval): """sets the cron interval of the user Args: user_id (int): id of the user cron_interval (String): Update interval in cron format => see https://crontab.guru/ for formatting Returns: int: status code """ with r.Session() as s: headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)} req = s.put(self.db_adress + "/user/setCron", json={"cron": cron_interval}, headers=headers) return req.status_code if __name__ == "__main__": print("This is a module for the telegram bot. It is not intended to be run directly.") handler = API_Handler("https://gruppe1.testsites.info/api", "bot@example.com", "bot") print(handler.token) keywords = handler.get_user_keywords(user_id = 1709356058) #user_id is currently mine (Linus) print(keywords) shares = handler.get_user_portfolio(user_id = 1709356058) print(handler.set_cron_interval(user_id = 1709356058, cron_interval = "0 0 * * *")) print(shares) sys.exit(1)