2022-05-12 11:30:56 +00:00
<!doctype html>
< html lang = "en" >
< head >
< meta charset = "utf-8" >
< meta name = "viewport" content = "width=device-width, initial-scale=1, minimum-scale=1" / >
< meta name = "generator" content = "pdoc 0.10.0" / >
< title > telegram_bot.api_handling.api_handler API documentation< / title >
< meta name = "description" content = "script for communicating with webservice to get data from database" / >
2022-09-18 11:05:35 +00:00
< link rel = "preload stylesheet" as = "style" href = "https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/13.0.0/sanitize.min.css" integrity = "sha512-y1dtMcuvtTMJc1yPgEqF0ZjQbhnc/bFhyvIyVNb9Zk5mIGtqVaAB1Ttl28su8AvFMOY0EwRbAe+HCLqj6W7/KA==" crossorigin >
< link rel = "preload stylesheet" as = "style" href = "https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/13.0.0/typography.min.css" integrity = "sha512-Y1DYSb995BAfxobCkKepB1BqJJTPrOp3zPL74AWFugHHmmdcvO+C48WLrUOlhGMc0QG7AE3f7gmvvcrmX2fDoA==" crossorigin >
2022-11-24 08:35:53 +00:00
< link rel = "stylesheet preload" as = "style" href = "https://cdnjs.cloudflare.com/ajax/libs/highlight.js/11.7.0/styles/github.min.css" crossorigin >
2022-05-12 11:30:56 +00:00
< style > : root { --highlight-color : #fe9 } . flex { display : flex !important } body { line-height : 1.5 em } # content { padding : 20 px } # sidebar { padding : 30 px ; overflow : hidden } # sidebar > * : last-child { margin-bottom : 2 cm } . http-server-breadcrumbs { font-size : 130 % ; margin : 0 0 15 px 0 } # footer { font-size : .75 em ; padding : 5 px 30 px ; border-top : 1 px solid #ddd ; text-align : right } # footer p { margin : 0 0 0 1 em ; display : inline-block } # footer p : last-child { margin-right : 30 px } h1 , h2 , h3 , h4 , h5 { font-weight : 300 } h1 { font-size : 2.5 em ; line-height : 1.1 em } h2 { font-size : 1.75 em ; margin : 1 em 0 .50 em 0 } h3 { font-size : 1.4 em ; margin : 25 px 0 10 px 0 } h4 { margin : 0 ; font-size : 105 % } h1 : target , h2 : target , h3 : target , h4 : target , h5 : target , h6 : target { background : var ( - - highlight - color ) ; padding : .2 em 0 } a { color : #058 ; text-decoration : none ; transition : color .3 s ease-in-out } a : hover { color : #e82 } . title code { font-weight : bold } h2 [ id ^ = "header-" ] { margin-top : 2 em } . ident { color : #900 } pre code { background : #f8f8f8 ; font-size : .8 em ; line-height : 1.4 em } code { background : #f2f2f1 ; padding : 1 px 4 px ; overflow-wrap : break-word } h1 code { background : transparent } pre { background : #f8f8f8 ; border : 0 ; border-top : 1 px solid #ccc ; border-bottom : 1 px solid #ccc ; margin : 1 em 0 ; padding : 1 ex } # http-server-module-list { display : flex ; flex-flow : column } # http-server-module-list div { display : flex } # http-server-module-list dt { min-width : 10 % } # http-server-module-list p { margin-top : 0 } . toc ul , # index { list-style-type : none ; margin : 0 ; padding : 0 } # index code { background : transparent } # index h3 { border-bottom : 1 px solid #ddd } # index ul { padding : 0 } # index h4 { margin-top : .6 em ; font-weight : bold } @ media ( min-width : 200ex ) { # index . two-column { column-count : 2 } } @ media ( min-width : 300ex ) { # index . two-column { column-count : 3 } } dl { margin-bottom : 2 em } dl dl : last-child { margin-bottom : 4 em } dd { margin : 0 0 1 em 3 em } # header-classes + dl > dd { margin-bottom : 3 em } dd dd { margin-left : 2 em } dd p { margin : 10 px 0 } . name { background : #eee ; font-weight : bold ; font-size : .85 em ; padding : 5 px 10 px ; display : inline-block ; min-width : 40 % } . name : hover { background : #e0e0e0 } dt : target . name { background : var ( - - highlight - color ) } . name > span : first-child { white-space : nowrap } . name . class > span : nth-child ( 2 ) { margin-left : .4 em } . inherited { color : #999 ; border-left : 5 px solid #eee ; padding-left : 1 em } . inheritance em { font-style : normal ; font-weight : bold } . desc h2 { font-weight : 400 ; font-size : 1.25 em } . desc h3 { font-size : 1 em } . desc dt code { background : inherit } . source summary , . git-link-div { color : #666 ; text-align : right ; font-weight : 400 ; font-size : .8 em ; text-transform : uppercase } . source summary > * { white-space : nowrap ; cursor : pointer } . git-link { color : inherit ; margin-left : 1 em } . source pre { max-height : 500 px ; overflow : auto ; margin : 0 } . source pre code { font-size : 12 px ; overflow : visible } . hlist { list-style : none } . hlist li { display : inline } . hlist li : after { content : ',\2002' } . hlist li : last-child : after { content : none } . hlist . hlist { display : inline ; padding-left : 1 em } img { max-width : 100 % } td { padding : 0 .5 em } . admonition { padding : .1 em .5 em ; margin-bottom : 1 em } . admonition-title { font-weight : bold } . admonition . note , . admonition . info , . admonition . important { background : #aef } . admonition . todo , . admonition . versionadded , . admonition . tip , . admonition . hint { background : #dfd } . admonition . warning , . admonition . versionchanged , . admonition . deprecated { background : #fd4 } . admonition . error , . admonition . danger , . admonition . caution { background : lightpink } < / style >
< style media = "screen and (min-width: 700px)" > @ media screen and ( min-width : 700px ) { # sidebar { width : 30 % ; height : 100 vh ; overflow : auto ; position : sticky ; top : 0 } # content { width : 70 % ; max-width : 100 ch ; padding : 3 em 4 em ; border-left : 1 px solid #ddd } pre code { font-size : 1 em } . item . name { font-size : 1 em } main { display : flex ; flex-direction : row-reverse ; justify-content : flex-end } . toc ul ul , # index ul { padding-left : 1.5 em } . toc > ul > li { margin-top : .5 em } } < / style >
< style media = "print" > @ media print { # sidebar h1 { page-break-before : always } . source { display : none } } @ media print { * { background : transparent !important ; color : #000 !important ; box-shadow : none !important ; text-shadow : none !important } a [ href ] : after { content : " (" attr ( href ) ")" ; font-size : 90 % } a [ href ] [ title ] : after { content : none } abbr [ title ] : after { content : " (" attr ( title ) ")" } . ir a : after , a [ href ^ = "javascript:" ] : after , a [ href ^ = "#" ] : after { content : "" } pre , blockquote { border : 1 px solid #999 ; page-break-inside : avoid } thead { display : table-header-group } tr , img { page-break-inside : avoid } img { max-width : 100 % !important } @ page { margin : 0 . 5cm } p , h2 , h3 { orphans : 3 ; widows : 3 } h1 , h2 , h3 , h4 , h5 , h6 { page-break-after : avoid } } < / style >
2022-11-24 08:35:53 +00:00
< script defer src = "https://cdnjs.cloudflare.com/ajax/libs/highlight.js/11.7.0/highlight.min.js" integrity = "sha512-bgHRAiTjGrzHzLyKOnpFvaEpGzJet3z4tZnXGjpsCcqOnAH6VGUx9frc5bcIhKTVLEiCO6vEhNAgx5jtLUYrfA==" crossorigin > < / script >
2022-05-12 11:30:56 +00:00
< script > window . addEventListener ( 'DOMContentLoaded' , ( ) => hljs . initHighlighting ( ) ) < / script >
< / head >
< body >
< main >
< article id = "content" >
< header >
< h1 class = "title" > Module < code > telegram_bot.api_handling.api_handler< / code > < / h1 >
< / header >
< section id = "section-intro" >
< p > script for communicating with webservice to get data from database< / p >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > " " "
script for communicating with webservice to get data from database
" " "
__author__ = " Florian Kellermann, Linus Eickhoff"
__date__ = " 10.05.2022"
__version__ = " 1.0.2"
__license__ = " None"
# side-dependencies: none
# Work in Progress
import os
import sys
import requests as r
from croniter import croniter # used for checking cron formatting
from dotenv import load_dotenv
load_dotenv() # loads environment vars
# note: for more information about the api visit swagger documentation on https://gruppe1.testsites.info/api/docs#/
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
get_user_portfolio(user_id): gets the portfolio of the user
set_portfolio(user_id, portfolio): sets the portfolio of the user
delete_portfolio(user_id, portfolio): deletes the portfolio of the user
set_cron_interval(user_id, interval): sets the cron interval of the user
set_admin(email, is_admin): sets the admin status of the user with the given email
" " "
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} # credentials for admin account that has all permissions to get and set data (in this case bot account)
with r.Session() as s: # open session
p = s.post(self.db_adress + " /user/login" , json=payload) # login to webservice
if p.status_code == 200:
self.token = p.json()[" data" ][' token' ] # store token for further authentication of requests
else:
print(" Error: " + str(p.status_code) + " invalid credentials" )
self.token = None
def reauthorize(self, email, password): # can be used if token expired
" " " set new credentials
Args:
email (string): email of the user
password (string): password of the user
Returns:
token (string): new token or None if not 200
Raises:
None
" " "
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): # max retries are used recursively if the request fails
" " " 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
Raises:
None
" " "
if max_retries < = 0:
return None
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)} # authorization is bot_token:user_id (user_id is the id of the user you want to get data from)
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) # if request fails try again recursively
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
Raises:
None
" " "
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
Raises:
None
" " "
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_json is a list of dictionaries
keywords.append(keyword[" keyword" ])
return keywords # will be empty if no keywords are set
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
Raises:
None
" " "
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
Raises:
None
" " "
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
Raises:
None
" " "
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[" isin" ]) # we only want the isin of the shares
return shares
else:
return self.get_user_shares(user_id, max_retries - 1)
def set_share(self, user_id, isin, comment):
" " " sets the share of the user
Args:
user_id (int): id of the user
isin (string): identifier of the share (standard is isin)
comment (string): comment of the share
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
req = s.post(self.db_adress + " /share" , json={" comment" : comment, " isin" : isin},
headers=headers) # set share by setting comment and isin, comment can be the real name of the share e.g. " Apple Inc."
return req.status_code
def delete_share(self, user_id, isin):
" " " deletes the share of the user
Args:
user_id (int): id of the user
isin (string): identifier of the share (standard is isin)
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
req = s.delete(self.db_adress + " /share" , json={" isin" : str(isin)}, headers=headers) # to delete a share only the isin is needed because it is unique, shares are not transactions!
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
Raises:
None
" " "
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 = req.json()[" data" ]
return transactions_dict
else:
return self.get_user_transactions(user_id, max_retries - 1)
def set_transaction(self, user_id, comment, isin, count, price, time):
" " " sets the transaction of the user
Args:
user_id (int): id of the user
comment (string): comment of the transaction
isin (string): isin of the transaction
count (float): count of the transaction
price (float): price of the transaction
time (string): time of the transaction formatted like e.g. " 2011-10-05T14:48:00.000Z"
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
time = time[:-3] + " Z" # remove last character and add Z to make it a valid date for db
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
transaction = {" comment" : str(comment), " count" : float(count), " isin" : str(isin), " price" : float(price),
" time" : str(time)} # set transaction as JSON with all the attributes needed according to Swagger docs
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
Raises:
None
" " "
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) # get portfolio as JSON
if req.status_code == 200:
portfolio_dict = req.json()[" data" ] # get the data of the JSON
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
Raises:
None
" " "
if not croniter.is_valid(cron_interval): # check if cron_interval is in valid format
print(" Error: Invalid cron format" )
return -1 # return error code -1 if invalid cron format
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
req = s.put(self.db_adress + " /user/setCron" , json={" cron" : str(cron_interval)}, headers=headers) # put not post (see swagger docs)
return req.status_code
def set_admin(self, email, is_admin):
" " " sets the admin of the user
Args:
email (string): email of the user
is_admin (bool): " true" if user should be Admin, " false" if not
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token} # only bot token is needed, user is chosen by email
req = s.put(self.db_adress + " /user/setAdmin" , json={" admin" : is_admin, " email" : str(email)}, headers=headers)
return req.status_code
if __name__ == " __main__" : # editable, just for basic on the go testing of new functions
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" , str(os.getenv(" BOT_EMAIL" )), str(os.getenv(" BOT_PASSWORD" ))) # get creds from env
print(handler.token)
keywords = handler.get_user_keywords(user_id=1709356058) # user_id here is currently mine (Linus)
print(keywords)
shares = handler.get_user_portfolio(user_id=1709356058)
print(" set cron with status: " + str(handler.set_cron_interval(user_id=1709356058, cron_interval=" 0 0 * * *" )))
user = handler.get_user(user_id=1709356058)
print(user)
all_users = handler.get_all_users()
admin_status = handler.set_admin(" test@test.com" , " true" )
print(admin_status)
print(all_users)
print(shares)
sys.exit(1)< / code > < / pre >
< / details >
< / section >
< section >
< / section >
< section >
< / section >
< section >
< / section >
< section >
< h2 class = "section-title" id = "header-classes" > Classes< / h2 >
< dl >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler" > < code class = "flex name class" >
< span > class < span class = "ident" > API_Handler< / span > < / span >
< span > (< / span > < span > db_adress, email, password)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > class for interacting with the api webservice< / p >
< h2 id = "attributes" > Attributes< / h2 >
< dl >
< dt > < strong > < code > db_adress< / code > < / strong > :  < code > string< / code > < / dt >
< dd > adress of the database< / dd >
< dt > < strong > < code > token< / code > < / strong > :  < code > string< / code > < / dt >
< dd > auth token for api< / dd >
< / dl >
< h2 id = "methods" > Methods< / h2 >
< p > 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
get_user_portfolio(user_id): gets the portfolio of the user
set_portfolio(user_id, portfolio): sets the portfolio of the user
delete_portfolio(user_id, portfolio): deletes the portfolio of the user
set_cron_interval(user_id, interval): sets the cron interval of the user
set_admin(email, is_admin): sets the admin status of the user with the given email< / p >
< p > initializes the API_Handler class< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > db_adress< / code > < / strong > :  < code > string< / code > < / dt >
< dd > adress of the database< / dd >
< dt > < strong > < code > email< / code > < / strong > :  < code > string< / code > < / dt >
< dd > email of the user< / dd >
< dt > < strong > < code > password< / code > < / strong > :  < code > string< / code > < / dt >
< dd > password of the user< / dd >
< / dl > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > 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
get_user_portfolio(user_id): gets the portfolio of the user
set_portfolio(user_id, portfolio): sets the portfolio of the user
delete_portfolio(user_id, portfolio): deletes the portfolio of the user
set_cron_interval(user_id, interval): sets the cron interval of the user
set_admin(email, is_admin): sets the admin status of the user with the given email
" " "
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} # credentials for admin account that has all permissions to get and set data (in this case bot account)
with r.Session() as s: # open session
p = s.post(self.db_adress + " /user/login" , json=payload) # login to webservice
if p.status_code == 200:
self.token = p.json()[" data" ][' token' ] # store token for further authentication of requests
else:
print(" Error: " + str(p.status_code) + " invalid credentials" )
self.token = None
def reauthorize(self, email, password): # can be used if token expired
" " " set new credentials
Args:
email (string): email of the user
password (string): password of the user
Returns:
token (string): new token or None if not 200
Raises:
None
" " "
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): # max retries are used recursively if the request fails
" " " 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
Raises:
None
" " "
if max_retries < = 0:
return None
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)} # authorization is bot_token:user_id (user_id is the id of the user you want to get data from)
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) # if request fails try again recursively
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
Raises:
None
" " "
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
Raises:
None
" " "
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_json is a list of dictionaries
keywords.append(keyword[" keyword" ])
return keywords # will be empty if no keywords are set
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
Raises:
None
" " "
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
Raises:
None
" " "
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
Raises:
None
" " "
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[" isin" ]) # we only want the isin of the shares
return shares
else:
return self.get_user_shares(user_id, max_retries - 1)
def set_share(self, user_id, isin, comment):
" " " sets the share of the user
Args:
user_id (int): id of the user
isin (string): identifier of the share (standard is isin)
comment (string): comment of the share
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
req = s.post(self.db_adress + " /share" , json={" comment" : comment, " isin" : isin},
headers=headers) # set share by setting comment and isin, comment can be the real name of the share e.g. " Apple Inc."
return req.status_code
def delete_share(self, user_id, isin):
" " " deletes the share of the user
Args:
user_id (int): id of the user
isin (string): identifier of the share (standard is isin)
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
req = s.delete(self.db_adress + " /share" , json={" isin" : str(isin)}, headers=headers) # to delete a share only the isin is needed because it is unique, shares are not transactions!
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
Raises:
None
" " "
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 = req.json()[" data" ]
return transactions_dict
else:
return self.get_user_transactions(user_id, max_retries - 1)
def set_transaction(self, user_id, comment, isin, count, price, time):
" " " sets the transaction of the user
Args:
user_id (int): id of the user
comment (string): comment of the transaction
isin (string): isin of the transaction
count (float): count of the transaction
price (float): price of the transaction
time (string): time of the transaction formatted like e.g. " 2011-10-05T14:48:00.000Z"
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
time = time[:-3] + " Z" # remove last character and add Z to make it a valid date for db
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
transaction = {" comment" : str(comment), " count" : float(count), " isin" : str(isin), " price" : float(price),
" time" : str(time)} # set transaction as JSON with all the attributes needed according to Swagger docs
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
Raises:
None
" " "
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) # get portfolio as JSON
if req.status_code == 200:
portfolio_dict = req.json()[" data" ] # get the data of the JSON
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
Raises:
None
" " "
if not croniter.is_valid(cron_interval): # check if cron_interval is in valid format
print(" Error: Invalid cron format" )
return -1 # return error code -1 if invalid cron format
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
req = s.put(self.db_adress + " /user/setCron" , json={" cron" : str(cron_interval)}, headers=headers) # put not post (see swagger docs)
return req.status_code
def set_admin(self, email, is_admin):
" " " sets the admin of the user
Args:
email (string): email of the user
is_admin (bool): " true" if user should be Admin, " false" if not
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token} # only bot token is needed, user is chosen by email
req = s.put(self.db_adress + " /user/setAdmin" , json={" admin" : is_admin, " email" : str(email)}, headers=headers)
return req.status_code< / code > < / pre >
< / details >
< h3 > Methods< / h3 >
< dl >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.delete_keyword" > < code class = "name flex" >
< span > def < span class = "ident" > delete_keyword< / span > < / span > (< span > self, user_id, keyword)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > deletes the keyword of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > keyword< / code > < / strong > :  < code > string< / code > < / dt >
< dd > keyword of the user< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > int< / code > < / dt >
< dd > status code< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > 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
Raises:
None
" " "
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< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.delete_share" > < code class = "name flex" >
< span > def < span class = "ident" > delete_share< / span > < / span > (< span > self, user_id, isin)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > deletes the share of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > isin< / code > < / strong > :  < code > string< / code > < / dt >
< dd > identifier of the share (standard is isin)< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > int< / code > < / dt >
< dd > status code< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > def delete_share(self, user_id, isin):
" " " deletes the share of the user
Args:
user_id (int): id of the user
isin (string): identifier of the share (standard is isin)
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
req = s.delete(self.db_adress + " /share" , json={" isin" : str(isin)}, headers=headers) # to delete a share only the isin is needed because it is unique, shares are not transactions!
return req.status_code< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.get_all_users" > < code class = "name flex" >
< span > def < span class = "ident" > get_all_users< / span > < / span > (< span > self, max_retries=10)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > gets all users< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > max_retries< / code > < / strong > :  < code > int< / code > < / dt >
< dd > max retries for the request< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > list< / code > < / dt >
< dd > list of users< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > 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
Raises:
None
" " "
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)< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.get_user" > < code class = "name flex" >
< span > def < span class = "ident" > get_user< / span > < / span > (< span > self, user_id, max_retries=10)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > gets the shares of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > max_retries< / code > < / strong > :  < code > int< / code > < / dt >
< dd > max retries for the request< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > json< / code > < / dt >
< dd > json of user infos< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > def get_user(self, user_id, max_retries=10): # max retries are used recursively if the request fails
" " " 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
Raises:
None
" " "
if max_retries < = 0:
return None
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)} # authorization is bot_token:user_id (user_id is the id of the user you want to get data from)
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) # if request fails try again recursively< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.get_user_keywords" > < code class = "name flex" >
< span > def < span class = "ident" > get_user_keywords< / span > < / span > (< span > self, user_id, max_retries=10)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > gets the keywords of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > max_retries< / code > < / strong > :  < code > int< / code > < / dt >
< dd > max retries for the request< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > list< / code > < / dt >
< dd > list of keywords< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > 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
Raises:
None
" " "
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_json is a list of dictionaries
keywords.append(keyword[" keyword" ])
return keywords # will be empty if no keywords are set
else:
return self.get_user_keywords(user_id, max_retries - 1)< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.get_user_portfolio" > < code class = "name flex" >
< span > def < span class = "ident" > get_user_portfolio< / span > < / span > (< span > self, user_id, max_retries=10)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > gets the portfolio of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > max_retries< / code > < / strong > :  < code > int< / code > < / dt >
< dd > max retries for the request< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > dict< / code > < / dt >
< dd > dictionary of portfolio< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > 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
Raises:
None
" " "
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) # get portfolio as JSON
if req.status_code == 200:
portfolio_dict = req.json()[" data" ] # get the data of the JSON
return portfolio_dict
else:
return self.get_user_portfolio(user_id, max_retries - 1)< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.get_user_shares" > < code class = "name flex" >
< span > def < span class = "ident" > get_user_shares< / span > < / span > (< span > self, user_id, max_retries=10)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > gets the shares of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > max_retries< / code > < / strong > :  < code > int< / code > < / dt >
< dd > max retries for the request< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > list< / code > < / dt >
< dd > list of shares< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > 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
Raises:
None
" " "
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[" isin" ]) # we only want the isin of the shares
return shares
else:
return self.get_user_shares(user_id, max_retries - 1)< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.get_user_transactions" > < code class = "name flex" >
< span > def < span class = "ident" > get_user_transactions< / span > < / span > (< span > self, user_id, max_retries=10)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > gets the transactions of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > max_retries< / code > < / strong > :  < code > int< / code > < / dt >
< dd > max retries for the request< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > dict< / code > < / dt >
< dd > dictionary of transactions< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > 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
Raises:
None
" " "
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 = req.json()[" data" ]
return transactions_dict
else:
return self.get_user_transactions(user_id, max_retries - 1)< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.reauthorize" > < code class = "name flex" >
< span > def < span class = "ident" > reauthorize< / span > < / span > (< span > self, email, password)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > set new credentials< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > email< / code > < / strong > :  < code > string< / code > < / dt >
< dd > email of the user< / dd >
< dt > < strong > < code > password< / code > < / strong > :  < code > string< / code > < / dt >
< dd > password of the user< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< p > token (string): new token or None if not 200< / p >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > def reauthorize(self, email, password): # can be used if token expired
" " " set new credentials
Args:
email (string): email of the user
password (string): password of the user
Returns:
token (string): new token or None if not 200
Raises:
None
" " "
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< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.set_admin" > < code class = "name flex" >
< span > def < span class = "ident" > set_admin< / span > < / span > (< span > self, email, is_admin)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > sets the admin of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > email< / code > < / strong > :  < code > string< / code > < / dt >
< dd > email of the user< / dd >
< dt > < strong > < code > is_admin< / code > < / strong > :  < code > bool< / code > < / dt >
< dd > "true" if user should be Admin, "false" if not< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > int< / code > < / dt >
< dd > status code< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > def set_admin(self, email, is_admin):
" " " sets the admin of the user
Args:
email (string): email of the user
is_admin (bool): " true" if user should be Admin, " false" if not
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token} # only bot token is needed, user is chosen by email
req = s.put(self.db_adress + " /user/setAdmin" , json={" admin" : is_admin, " email" : str(email)}, headers=headers)
return req.status_code< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.set_cron_interval" > < code class = "name flex" >
< span > def < span class = "ident" > set_cron_interval< / span > < / span > (< span > self, user_id, cron_interval)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > sets the cron interval of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > cron_interval< / code > < / strong > :  < code > String< / code > < / dt >
< dd > Update interval in cron format => see < a href = "https://crontab.guru/" > https://crontab.guru/< / a > for formatting< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > int< / code > < / dt >
< dd > status code< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > 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
Raises:
None
" " "
if not croniter.is_valid(cron_interval): # check if cron_interval is in valid format
print(" Error: Invalid cron format" )
return -1 # return error code -1 if invalid cron format
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
req = s.put(self.db_adress + " /user/setCron" , json={" cron" : str(cron_interval)}, headers=headers) # put not post (see swagger docs)
return req.status_code< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.set_keyword" > < code class = "name flex" >
< span > def < span class = "ident" > set_keyword< / span > < / span > (< span > self, user_id, keyword)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > sets the keyword of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > keyword< / code > < / strong > :  < code > int< / code > < / dt >
< dd > keyword of the user< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > int< / code > < / dt >
< dd > status code< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > 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
Raises:
None
" " "
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< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.set_share" > < code class = "name flex" >
< span > def < span class = "ident" > set_share< / span > < / span > (< span > self, user_id, isin, comment)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > sets the share of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > isin< / code > < / strong > :  < code > string< / code > < / dt >
< dd > identifier of the share (standard is isin)< / dd >
< dt > < strong > < code > comment< / code > < / strong > :  < code > string< / code > < / dt >
< dd > comment of the share< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > int< / code > < / dt >
< dd > status code< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > def set_share(self, user_id, isin, comment):
" " " sets the share of the user
Args:
user_id (int): id of the user
isin (string): identifier of the share (standard is isin)
comment (string): comment of the share
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
req = s.post(self.db_adress + " /share" , json={" comment" : comment, " isin" : isin},
headers=headers) # set share by setting comment and isin, comment can be the real name of the share e.g. " Apple Inc."
return req.status_code< / code > < / pre >
< / details >
< / dd >
< dt id = "telegram_bot.api_handling.api_handler.API_Handler.set_transaction" > < code class = "name flex" >
< span > def < span class = "ident" > set_transaction< / span > < / span > (< span > self, user_id, comment, isin, count, price, time)< / span >
< / code > < / dt >
< dd >
< div class = "desc" > < p > sets the transaction of the user< / p >
< h2 id = "args" > Args< / h2 >
< dl >
< dt > < strong > < code > user_id< / code > < / strong > :  < code > int< / code > < / dt >
< dd > id of the user< / dd >
< dt > < strong > < code > comment< / code > < / strong > :  < code > string< / code > < / dt >
< dd > comment of the transaction< / dd >
< dt > < strong > < code > isin< / code > < / strong > :  < code > string< / code > < / dt >
< dd > isin of the transaction< / dd >
< dt > < strong > < code > count< / code > < / strong > :  < code > float< / code > < / dt >
< dd > count of the transaction< / dd >
< dt > < strong > < code > price< / code > < / strong > :  < code > float< / code > < / dt >
< dd > price of the transaction< / dd >
< dt > < strong > < code > time< / code > < / strong > :  < code > string< / code > < / dt >
< dd > time of the transaction formatted like e.g. "2011-10-05T14:48:00.000Z"< / dd >
< / dl >
< h2 id = "returns" > Returns< / h2 >
< dl >
< dt > < code > int< / code > < / dt >
< dd > status code< / dd >
< / dl >
< h2 id = "raises" > Raises< / h2 >
< p > None< / p > < / div >
< details class = "source" >
< summary >
< span > Expand source code< / span >
< / summary >
< pre > < code class = "python" > def set_transaction(self, user_id, comment, isin, count, price, time):
" " " sets the transaction of the user
Args:
user_id (int): id of the user
comment (string): comment of the transaction
isin (string): isin of the transaction
count (float): count of the transaction
price (float): price of the transaction
time (string): time of the transaction formatted like e.g. " 2011-10-05T14:48:00.000Z"
Returns:
int: status code
Raises:
None
" " "
with r.Session() as s:
time = time[:-3] + " Z" # remove last character and add Z to make it a valid date for db
headers = {' Authorization' : ' Bearer ' + self.token + " :" + str(user_id)}
transaction = {" comment" : str(comment), " count" : float(count), " isin" : str(isin), " price" : float(price),
" time" : str(time)} # set transaction as JSON with all the attributes needed according to Swagger docs
req = s.post(self.db_adress + " /transaction" , json=transaction, headers=headers)
return req.status_code< / code > < / pre >
< / details >
< / dd >
< / dl >
< / dd >
< / dl >
< / section >
< / article >
< nav id = "sidebar" >
< h1 > Index< / h1 >
< div class = "toc" >
< ul > < / ul >
< / div >
< ul id = "index" >
< li > < h3 > Super-module< / h3 >
< ul >
< li > < code > < a title = "telegram_bot.api_handling" href = "index.html" > telegram_bot.api_handling< / a > < / code > < / li >
< / ul >
< / li >
< li > < h3 > < a href = "#header-classes" > Classes< / a > < / h3 >
< ul >
< li >
< h4 > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler" href = "#telegram_bot.api_handling.api_handler.API_Handler" > API_Handler< / a > < / code > < / h4 >
< ul class = "" >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.delete_keyword" href = "#telegram_bot.api_handling.api_handler.API_Handler.delete_keyword" > delete_keyword< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.delete_share" href = "#telegram_bot.api_handling.api_handler.API_Handler.delete_share" > delete_share< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.get_all_users" href = "#telegram_bot.api_handling.api_handler.API_Handler.get_all_users" > get_all_users< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.get_user" href = "#telegram_bot.api_handling.api_handler.API_Handler.get_user" > get_user< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.get_user_keywords" href = "#telegram_bot.api_handling.api_handler.API_Handler.get_user_keywords" > get_user_keywords< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.get_user_portfolio" href = "#telegram_bot.api_handling.api_handler.API_Handler.get_user_portfolio" > get_user_portfolio< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.get_user_shares" href = "#telegram_bot.api_handling.api_handler.API_Handler.get_user_shares" > get_user_shares< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.get_user_transactions" href = "#telegram_bot.api_handling.api_handler.API_Handler.get_user_transactions" > get_user_transactions< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.reauthorize" href = "#telegram_bot.api_handling.api_handler.API_Handler.reauthorize" > reauthorize< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.set_admin" href = "#telegram_bot.api_handling.api_handler.API_Handler.set_admin" > set_admin< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.set_cron_interval" href = "#telegram_bot.api_handling.api_handler.API_Handler.set_cron_interval" > set_cron_interval< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.set_keyword" href = "#telegram_bot.api_handling.api_handler.API_Handler.set_keyword" > set_keyword< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.set_share" href = "#telegram_bot.api_handling.api_handler.API_Handler.set_share" > set_share< / a > < / code > < / li >
< li > < code > < a title = "telegram_bot.api_handling.api_handler.API_Handler.set_transaction" href = "#telegram_bot.api_handling.api_handler.API_Handler.set_transaction" > set_transaction< / a > < / code > < / li >
< / ul >
< / li >
< / ul >
< / li >
< / ul >
< / nav >
< / main >
< footer id = "footer" >
< p > Generated by < a href = "https://pdoc3.github.io/pdoc" title = "pdoc: Python API documentation generator" > < cite > pdoc< / cite > 0.10.0< / a > .< / p >
< / footer >
< / body >
< / html >
