Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# The existing access_token is expired
# Attempt to refresh the token
try:
new_access_info = self.cli.refresh_access_token(
client_id=STRAVA_CLIENT_ID,
client_secret=STRAVA_CLIENT_SECRET,
refresh_token=access_info.get("refresh_token"))
self.access_token = json.dumps(new_access_info)
try:
session.commit()
except Exception:
log.exception("postgres error")
self.cli = stravalib.Client(
access_token=new_access_info.get("access_token"),
rate_limiter=(lambda x=None: None)
)
except Exception:
log.exception("%s token refresh fail", self)
self.cli = None
else:
elapsed = round(time.time() - start, 2)
log.debug("%s token refresh elapsed=%s", self, elapsed)
return self.cli
log.exception("error inserting event %s", event)
@classmethod
def log_request(cls, flask_request_object, **args):
req = flask_request_object
args.update({
"ip": req.access_route[-1],
"agent": vars(req.user_agent),
})
cls.new_event(**args)
class Webhooks(object):
name = "subscription"
client = stravalib.Client()
credentials = {
"client_id": STRAVA_CLIENT_ID,
"client_secret": STRAVA_CLIENT_SECRET
}
@classmethod
def create(cls, callback_url):
try:
subs = cls.client.create_subscription(
callback_url=callback_url,
**cls.credentials
)
except Exception as e:
log.exception("error creating subscription")
return dict(error=str(e))
def main(action, filename):
if action != "DOWNLOAD":
return 0
try:
with open(STRAVA_CREDENTIALS_FILE, 'r') as f:
access_token = f.read().strip(' \t\n\r')
except FileNotFoundError:
print('No Strava credentials provided.')
print('You first need to run the script to fetch the credentials')
print('./40-upload_to_strava.py')
return -1
try:
client = Client(access_token=access_token)
print('Uploading {}: '.format(os.path.basename(filename)), end='')
with open(filename, 'rb') as f:
upload = client.upload_activity(
activity_file=f,
data_type='fit',
private=STRAVA_UPLOAD_PRIVATE,
)
except (ActivityUploadFailed, FileNotFoundError) as err:
print('FAILED')
print('Reason:', err)
return -1
print('SUCCESS')
return 0
flash("Error: {}".format(request.args.get("error")))
return redirect(state or url_for("splash"))
scope = request.args.get("scope")
if "activity:read" not in scope:
# We need to be able to read the user's activities
return redirect(url_for("authorize", state=state))
if current_user.is_anonymous:
args = dict(
code=request.args.get("code"),
client_id=app.config["STRAVA_CLIENT_ID"],
client_secret=app.config["STRAVA_CLIENT_SECRET"]
)
client = stravalib.Client()
try:
access_info = client.exchange_code_for_token(**args)
# access_info is a dict containing the access_token,
# date of expire, and a refresh token
user_data = Users.strava_user_data(
access_info=access_info
)
assert user_data
new_user = "" if Users.get(user_data["id"]) else " NEW USER"
user = Users.add_or_update(**user_data)
assert user
def logged_in():
"""
Method called by Strava (redirect) that includes parameters.
- state
- code
- error
"""
error = request.args.get("error")
state = request.args.get("state")
if error:
return render_template(
"login_error.html", error=error, competition_title=config.COMPETITION_TITLE
)
else:
code = request.args.get("code")
client = Client()
token_dict = client.exchange_code_for_token(
client_id=config.STRAVA_CLIENT_ID,
client_secret=config.STRAVA_CLIENT_SECRET,
code=code,
)
# Use the now-authenticated client to get the current athlete
strava_athlete = client.get_athlete()
athlete_model = data.update_athlete_auth(strava_athlete, token_dict)
if not athlete_model:
return render_template(
"login_error.html",
error="ATHLETE_NOT_FOUND",
competition_title=config.COMPETITION_TITLE,
)
def authorize():
state = request.args.get("state")
redirect_uri = url_for('auth_callback', _external=True)
client = stravalib.Client()
auth_url = client.authorization_url(
client_id=app.config["STRAVA_CLIENT_ID"],
redirect_uri=redirect_uri,
approval_prompt="force",
scope=["read", "activity:read", "activity:read_all"],
state=state
)
return redirect(auth_url)
def login():
c = Client()
url = c.authorization_url(
client_id=config.STRAVA_CLIENT_ID,
redirect_uri=url_for(".logged_in", _external=True),
approval_prompt="auto",
scope=["read_all", "activity:read_all", "profile:read_all"],
)
return render_template(
"login.html", authorize_url=url, competition_title=config.COMPETITION_TITLE
)
def webhook_challenge():
client = Client()
strava_request = {
k: request.args.get(k)
for k in ("hub.challenge", "hub.mode", "hub.verify_token")
}
log.info("Webhook challenge: {}".format(strava_request))
challenge_resp = client.handle_subscription_callback(
strava_request, verify_token=config.STRAVA_VERIFY_TOKEN
)
return jsonify(challenge_resp)
def import_activities(db, user, limit=1):
count = 0
msg = "importing activities from Strava..."
# logging.info(msg)
yield msg + "\n"
token = user.strava_access_token
client = stravalib.Client(access_token=token)
activities = client.get_activities(limit=limit)
while True:
try:
a = activities.next()
except StopIteration:
msg = "done importing {} activities.".format(count)
# logging.info(msg)
yield msg + "\n"
return
except Exception as e:
yield str(e)
return
count += 1
flash("Error: {}".format(request.args.get("error")))
return redirect(state or url_for("splash"))
scope = request.args.get("scope")
# log.debug("scope: {}".format(scope))
if "activity:read" not in scope:
# We need to be able to read the user's activities
return redirect(url_for("authorize", state=state))
if current_user.is_anonymous:
args = {"code": request.args.get("code"),
"client_id": app.config["STRAVA_CLIENT_ID"],
"client_secret": app.config["STRAVA_CLIENT_SECRET"]}
client = stravalib.Client()
try:
access_info = client.exchange_code_for_token(**args)
except Exception as e:
log.error("authorization error:\n{}".format(e))
flash(str(e))
return redirect(state)
# log.debug("got code exchange response: {}".format(access_info))
access_info_string = json.dumps(access_info)
user_data = Users.strava_data_from_token(access_info_string)
# log.debug("user data: {}".format(user_data))