TelegramAktienBot/telegram_bot/api_handler.py

"""
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 communicating with webservice to get data from database
    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)