Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_init(self, loop):
session = aiohttp.ClientSession()
fpl = FPL(session)
assert fpl.session is session
keys = [
"events",
"game_settings",
"phases",
"teams",
"elements",
"element_types",
"element_stats",
"total_players",
"current_gameweek",
]
assert all([hasattr(fpl, key) for key in keys])
assert all([isinstance(getattr(fpl, key), dict) for key in keys[:-3]])
assert isinstance(getattr(fpl, keys[-3]), list)
assert all([isinstance(getattr(fpl, key), int) for key in keys[-2:]])
def __init__(self, config, session):
self.config = config
self.database = client.fpl
self.fpl = FPL(session)
self.reddit = praw.Reddit(
client_id=config.get("CLIENT_ID"),
client_secret=config.get("CLIENT_SECRET"),
password=config.get("PASSWORD"),
user_agent=config.get("USER_AGENT"),
username=config.get("USERNAME"))
self.subreddit = self.reddit.subreddit(self.config.get("SUBREDDIT"))
async def main(config):
"""Returns a list of players whose price has changed since the last
time the database was updated.
"""
database = client.fpl
async with aiohttp.ClientSession() as session:
fpl = FPL(session)
fpl_bot = FPLBot(config, session)
new_players = await fpl.get_players(include_summary=True)
for new_player in new_players:
old_player = database.players.find_one({"id": new_player.id})
# New player has been added to the game
if not old_player:
continue
if (old_player["now_cost"] > new_player.now_cost or
old_player["now_cost"] < new_player.now_cost):
await fpl_bot.post_price_changes()
async def get_picks(team):
"""Returns a list of players with the necessary information to format the
team's formation properly.
"""
player_ids = [player["element"] for player in team]
async with aiohttp.ClientSession() as session:
fpl = FPL(session)
players = await fpl.get_players(player_ids)
for player_data in team:
for player in players:
if player_data["element"] != player.id:
continue
player.role = ""
player.event_points = (player.event_points *
player_data["multiplier"])
player.team_position = player_data["position"]
player.is_captain = player_data["is_captain"]
if player.is_captain:
player.role = " (C)"
async def update_players():
"""Updates all players in the database."""
logger.info("Updating FPL players in database.")
async with aiohttp.ClientSession() as session:
fpl = FPL(session)
players = await fpl.get_players(include_summary=True, return_json=True)
for player in players:
player["team"] = team_converter(player["team"])
requests = [ReplaceOne({"id": player["id"]}, player, upsert=True)
for player in players]
database.players.bulk_write(requests)
create_text_indexes()
logger.info("Adding Understat data to players in database.")
understat_players = await get_understat_players()
for player in understat_players:
# Only update FPL player with desired attributes
understat_attributes = {
attribute: value for attribute, value in player.items()
async def picks(user_id):
"""Echoes a user's picks to the terminal."""
async with aiohttp.ClientSession() as session:
fpl = FPL(session)
user = await fpl.get_user(user_id)
await format_picks(user)
async def myteam(user_id, email, password):
"""Echoes a logged in user's team to the terminal."""
if isinstance(password, HiddenPassword):
password = password.password
async with aiohttp.ClientSession() as session:
fpl = FPL(session)
await fpl.login(email, password)
try:
user = await fpl.get_user(user_id)
await format_myteam(user)
except KeyError:
raise click.BadParameter("email address or password.")