Skip to content
Foldr Docs

Worked example

A short client in Python that does the whole round trip: sign in with OAuth and PKCE, list the shares the person can write to, upload a file, copy it (asking for an operation, and waiting for it), and handle the answers a client has to expect. It uses only the requests library.

Before you run it, register an app as described in Getting access (an administrator adds it under Settings > Connected apps, with a redirect URL, and allows files.write in its policy) and set:

Terminal window
export FOLDR_SITE=https://files.example.com
export FOLDR_CLIENT_ID=...
export FOLDR_CLIENT_SECRET=...        # leave unset for a public app
export FOLDR_REDIRECT_URI=https://reports.acme.example/foldr/callback
export FOLDR_SHARE=...                # optional: the id of the share the app may write to
python foldr_example.py

The script prints a sign-in address, you open it in a browser and agree, and you paste back the address Foldr sent you to (the one that begins with your redirect URL and carries ?code=...). A real app would run a small web server on the redirect URL instead and read the code from the request.

"""A small Foldr API v1 client: sign in with OAuth (PKCE), list shares,
upload a file, copy a folder and wait for the copy to finish."""


import base64
import hashlib
import os
import secrets
import time
import uuid
from urllib.parse import parse_qs, urlencode, urlparse


import requests


SITE = os.environ["FOLDR_SITE"]                          # https://files.example.com
CLIENT_ID = os.environ["FOLDR_CLIENT_ID"]
CLIENT_SECRET = os.environ.get("FOLDR_CLIENT_SECRET")    # a confidential app only
REDIRECT_URI = os.environ["FOLDR_REDIRECT_URI"]
API = SITE + "/api/v1"




# --- OAuth: authorisation code with PKCE ---------------------------------


def make_pkce():
    verifier = secrets.token_urlsafe(48)                 # 64 characters of A-Za-z0-9_-
    digest = hashlib.sha256(verifier.encode("ascii")).digest()
    challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
    return verifier, challenge




def authorize_url(challenge, state, scope):
    return SITE + "/oauth/authorize?" + urlencode({
        "response_type": "code",
        "client_id": CLIENT_ID,
        "redirect_uri": REDIRECT_URI,
        "scope": scope,
        "state": state,
        "code_challenge": challenge,
        "code_challenge_method": "S256",
        "resource": API,
    })




def token_request(**form):
    auth = (CLIENT_ID, CLIENT_SECRET) if CLIENT_SECRET else None
    if auth is None:
        form["client_id"] = CLIENT_ID
    r = requests.post(SITE + "/oauth/token", data=form, auth=auth, timeout=30)
    if r.status_code != 200:
        raise RuntimeError("token endpoint: " + r.text)  # e.g. {"error":"invalid_grant"}
    return r.json()




def exchange_code(code, verifier):
    return token_request(grant_type="authorization_code", code=code,
                         redirect_uri=REDIRECT_URI, code_verifier=verifier)




# --- The API client -------------------------------------------------------


class FoldrError(Exception):
    """A problem document (RFC 9457). Branch on .type, never on the text."""


    def __init__(self, response):
        try:
            problem = response.json()
        except ValueError:
            problem = {}
        self.status = response.status_code
        self.type = str(problem.get("type", "")).rsplit("/", 1)[-1]
        self.detail = problem.get("detail")
        super().__init__(f"{self.status} {self.type}")




class Foldr:
    def __init__(self, tokens, save_tokens=lambda tokens: None):
        self.tokens = tokens
        self.save_tokens = save_tokens
        self.http = requests.Session()


    def refresh(self):
        # The refresh token is used once: keep the new one before anything else.
        self.tokens = token_request(grant_type="refresh_token",
                                    refresh_token=self.tokens["refresh_token"])
        self.save_tokens(self.tokens)


    def call(self, method, path, ok=(200, 201, 202, 204), **kwargs):
        headers = kwargs.pop("headers", {})
        url = path if path.startswith("http") else API + path   # an operation's Location is a full URL
        for attempt in range(4):
            headers["Authorization"] = "Bearer " + self.tokens["access_token"]
            r = self.http.request(method, url, headers=headers, timeout=120,
                                  allow_redirects=False, **kwargs)
            if r.status_code == 401 and attempt == 0:
                self.refresh()                           # expired: refresh once, then retry
                continue
            if r.status_code in (429, 503):              # flow control, or a share briefly away
                wait = int(r.headers.get("Retry-After", 5))
                if wait > 120:                           # a daily limit: not for sleeping through
                    break
                time.sleep(wait)
                continue
            break
        if r.status_code not in ok:
            raise FoldrError(r)
        return r


    # Reading


    def pages(self, path, params=None):
        params = dict(params or {})
        while True:
            page = self.call("GET", path, params=params).json()
            yield from page["items"]
            if not page["next_cursor"]:
                return
            params["cursor"] = page["next_cursor"]       # opaque: send it back as it came


    # Writing


    def upload(self, share_id, path, data: bytes):
        r = self.call("PUT", f"/shares/{share_id}/files", params={"path": path},
                      headers={"If-None-Match": "*", "Idempotency-Key": str(uuid.uuid4()),
                               "Content-Type": "application/octet-stream"},
                      data=data)
        return r.json()                                  # read its "path": it may have been renamed


    def copy(self, share_id, path, to_share, to_path):
        r = self.call("POST", f"/shares/{share_id}/copy",
                      headers={"Idempotency-Key": str(uuid.uuid4()), "Prefer": "respond-async"},
                      json={"path": path, "to_share": to_share, "to_path": to_path})
        return self.wait(r) if r.status_code == 202 else r.json()


    def wait(self, response):
        """Poll an operation until it is no longer 202; return its result."""
        location = response.headers["Location"]
        delay = int(response.headers.get("Retry-After", 5))
        while True:
            time.sleep(delay)
            # Through call(): a refresh on 401 and a wait on 429/503 apply to polling too.
            r = self.call("GET", location, ok=(200, 202))
            if r.status_code == 202:
                delay = int(r.headers.get("Retry-After", 5))
                continue
            return r.json()["result"]                    # any other status raised: the problem the operation ended with




def main():
    verifier, challenge = make_pkce()
    state = secrets.token_urlsafe(16)
    print("Open this in a browser, sign in and agree:\n")
    print(authorize_url(challenge, state, "foldr.read offline_access files.write"))
    redirected = input("\nPaste the address you were sent back to: ")
    query = parse_qs(urlparse(redirected).query)
    if query.get("state") != [state]:
        raise SystemExit("state does not match: stop")
    if query.get("iss") != [SITE]:
        raise SystemExit("iss is not the site this app means to talk to: stop")
    foldr = Foldr(exchange_code(query["code"][0], verifier))
    print("granted:", foldr.tokens["scope"])


    # read_only describes the person's own permission, not the app's policy: an
    # administrator may have limited the app to named shares, and a share outside
    # them answers 404. A real app is configured with its destination share.
    shares = [s for s in foldr.pages("/shares") if not s["read_only"] and s["reachable"]]
    wanted = os.environ.get("FOLDR_SHARE")               # the share id the administrator approved
    if wanted:
        shares = [s for s in shares if s["id"] == wanted]
    if not shares:
        raise SystemExit("no share this person can write to (and the app may use) was found")
    share = shares[0]
    print("using share", share["name"])


    item = foldr.upload(share["id"], "/hello.txt", b"Hello from the API\n")
    print("uploaded", item["path"], item["size"], "bytes")


    try:
        copy = foldr.copy(share["id"], item["path"], share["id"], "/hello copy.txt")
        print("copied to", copy["path"])
    except FoldrError as e:
        if e.type in ("legal-hold", "checked-out"):
            print("not now:", e.type)
        else:
            raise




if __name__ == "__main__":
    main()

What to notice

Section titled “What to notice”
  • Everything goes through one call method, so the rules that apply to every request live in one place: the bearer header, one refresh on 401, and waiting out a short Retry-After on 429 and 503. It stops retrying after a few attempts and then raises. A Retry-After of more than two minutes is a daily limit (daily-write-limit or destructive-budget, which run to midnight UTC), so it raises at once for the caller to tell the user rather than sleeping.
  • The refresh token is saved the moment it changes (save_tokens). It is used once; losing the new one means the person has to sign in again.
  • Paging is a generator that hands back next_cursor untouched.
  • The upload sends If-None-Match: * and a fresh Idempotency-Key, takes its data as bytes (so a retry can send it again), and reads the path from the answer rather than assuming the name it asked for. For a big file, give requests an open file object instead, and don’t retry from inside call: reopen the file and start again.
  • A copy is always asked for as an operation (Prefer: respond-async), so there’s one path whatever the size: a 202 is followed to its end with wait, which sleeps for Retry-After between polls and polls through call, so a refresh or a short throttle does not end the wait.
  • Errors are read by type, not by text. FoldrError.type is the last part of the problem’s type URL (legal-hold, checked-out, not-found, …). The try at the end shows the usual shape: some problems are “not now” answers to show a person, and the rest you re-raise.
  • A key per request. The script makes a new key for every new request. If you retry a request yourself after a timeout, send the same key, so Foldr can recognise it.

What this example leaves out, and a real client shouldn’t: storing tokens somewhere safe, checking scope in the token response before relying on files.write, a place to show a user the daily-limit problems it raises, and revoking the token (RFC 7009) when the person disconnects the app on your side.