How to use the stravalib.Client function in stravalib

To help you get started, weโ€™ve selected a few stravalib examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github ebrensi / heatflask / heatflask / models.py View on Github external
# 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
github ebrensi / heatflask / heatflask / models.py View on Github external
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))
github Tigge / antfs-cli / scripts / 40-upload_to_strava.py View on Github external
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
github ebrensi / heatflask / heatflask / routes.py View on Github external
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
github freezingsaddles / freezing-web / freezing / web / views / general.py View on Github external
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,
            )
github ebrensi / heatflask / heatflask.py View on Github external
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)
github freezingsaddles / freezing-web / freezing / web / views / general.py View on Github external
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
    )
github freezingsaddles / freezing-web / freezing / web / views / general.py View on Github external
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)
github ebrensi / heatflask / stravaimport.py View on Github external
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
github ebrensi / heatflask / heatflask.py View on Github external
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))