""" 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: 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_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 ' + 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", "bot@example.com", "bot") print(handler.token) keywords = handler.get_user_keywords(user_id = 1709356058) #user_id is currently mine (Linus) print(keywords) sys.exit(1)