Coverage for betbot/api/ApiConnection.py : 20%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""LICENSE
2Copyright 2020 Hermann Krumrey <hermann@krumreyh.com>
4This file is part of betbot.
6betbot is free software: you can redistribute it and/or modify
7it under the terms of the GNU General Public License as published by
8the Free Software Foundation, either version 3 of the License, or
9(at your option) any later version.
11betbot is distributed in the hope that it will be useful,
12but WITHOUT ANY WARRANTY; without even the implied warranty of
13MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14GNU General Public License for more details.
16You should have received a copy of the GNU General Public License
17along with betbot. If not, see <http://www.gnu.org/licenses/>.
18LICENSE"""
20import os
21import json
22import requests
23import logging
24from typing import List, Dict, Optional, Any, Tuple
25from base64 import b64encode
26from betbot.api.Bet import Bet
27from betbot.api.Match import Match
30class ApiConnection:
31 """
32 Class that handles API calls to the bundesliga-tippspiel instance
33 """
35 def __init__(self, username: str, password: str, url: str):
36 """
37 Intitializes the API connection
38 :param username: The username on bundesliga-tippspiel
39 :param password: The password for that user
40 :param url: The base url to the bundesliga-tippspiel instance
41 """
42 self.logger = logging.getLogger(__name__)
43 self.username = username
44 self.password = password
45 self.url = os.path.join(url, "api/v3") + "/"
46 self.api_key = ""
47 self.login()
48 self.logger.debug("Initialized")
50 @property
51 def auth_headers(self) -> Dict[str, str]:
52 """
53 :return: Authorization headers for API calls
54 """
55 return {"Authorization": "Basic " + self.api_key}
57 def login(self) -> bool:
58 """
59 Retrieves an API Key using a username and password if no key
60 has been retrieved yet or if the existing key has expired
61 :return: True if the login was successfulle
62 """
63 if not self.authorized():
64 self.logger.info("Requesting new API Key")
65 creds = {"username": self.username, "password": self.password}
66 data = self.execute_api_call("key", "POST", False, creds)
68 if data["status"] != "ok":
69 self.logger.warning("Login attempt failed")
70 return False
71 else:
72 api_key = data["data"]["api_key"]
73 encoded = b64encode(api_key.encode("utf-8"))
74 self.api_key = encoded.decode("utf-8")
75 self.logger.info("Login successful")
76 return True
77 else:
78 return True
80 def authorized(self) -> bool:
81 """
82 Checks if the stored API key is valid
83 :return: True if valid, False if not (for example because it expired)
84 """
85 self.logger.debug("Authorization Check")
86 data = self.execute_api_call("authorize", "GET", True)
87 return data["status"] == "ok"
89 def logout(self):
90 """
91 Logs out the bot by deleting the API key
92 :return: None
93 """
94 self.execute_api_call(
95 "key", "DELETE", json_data={"api_key": self.api_key}
96 )
97 self.logger.info("Logging out.")
99 def get_active_leagues(self) -> List[Tuple[str, int]]:
100 """
101 :return: A list of tuples of leagues and seasons of active leagues
102 """
103 leagues = self.execute_api_call(
104 "leagues", "GET", True
105 )["data"]["leagues"]
106 newest_season = max([x[1] for x in leagues])
107 return [x for x in leagues if x[1] == newest_season]
109 def get_league_table(self, season: str, league: int) -> List[str]:
110 """
111 Retrieves the current league table order
112 :param season: The league for which to retrieve the league table
113 :param league: The season for which to retrieve the league table
114 :return: The team abbreviations in order of their league rankings
115 """
116 league_table = self.execute_api_call(
117 f"league_table/{season}/{league}", "GET", True
118 )["data"]["league_table"]
119 return [x[1]["abbreviation"] for x in league_table]
121 def get_current_matchday_matches(
122 self, league: str, season: int
123 ) -> List[Match]:
124 """
125 Retrieves a list of matches for the current matchday
126 :param league: The league to retrieve matches for
127 :param season: The season to retrieve matches for
128 :return: The list of matches
129 """
130 self.logger.debug("Getting current matches")
131 match_data = self.execute_api_call(
132 f"matchday/{league}/{season}", "GET", True
133 )["data"]["matches"]
134 return [Match.from_json(x) for x in match_data]
136 def place_bets(self, bets: List[Bet]):
137 """
138 Places a list of bets
139 :param bets: The bets to place
140 :return: None
141 """
142 for bet in bets:
143 self.logger.info(f"Placing bet: "
144 f"{bet.match.home_team} VS {bet.match.away_team}:"
145 f" {bet.home_score}:{bet.away_score}")
147 bet_dicts = [bet.to_dict() for bet in bets]
148 data = self.execute_api_call(
149 "place_bets", "PUT", True, {"bets": bet_dicts}
150 )
151 if data["status"] != "ok":
152 self.logger.error("Failed to place bets")
153 else:
154 placed = len(data["data"]["placed"])
155 self.logger.info(f"Placed {placed} bets (user:{self.username})")
157 def execute_api_call(
158 self,
159 endpoint: str,
160 method: str,
161 authorization_required: bool = False,
162 json_data: Optional[Dict[str, Any]] = None,
163 ) -> Dict[str, Any]:
164 """
165 Executes an API call
166 :param endpoint: The API endpoint
167 :param method: The request method
168 :param authorization_required: Whether authoirzation is required
169 :param json_data: The JSON data to send
170 :return: The response JSON
171 """
172 if authorization_required and endpoint != "authorize":
173 logged_in = self.login()
174 if not logged_in:
175 return {"status": "error"}
177 extras = {}
178 if authorization_required:
179 extras["headers"] = self.auth_headers
180 if json_data is not None:
181 extras["json"] = json_data
183 api_url = self.url + endpoint
184 resp = requests.request(method, api_url, **extras) # type: ignore
185 return json.loads(resp.text)