Child Item

How do I upload file using Python

I've checked here and over at StackOverflow but couldn't find a solution. Nothing works.

It says that I need refresh token (I have no clue where to get it) and an app key.

import pathlib
import dropbox

dropbox_access_token="my_token"
app_key="my_app_key"

folder = pathlib.Path(".")
filename = "img.jpg"
filepath = folder / filename

target = "/app_folder_dropbox/"
targetfile = target + filename

d = dropbox.Dropbox(dropbox_access_token, app_key=app_key)

with filepath.open("rb") as f:
    meta = d.files_upload(f.read(), targetfile, mode=dropbox.files.WriteMode("overwrite"))

Is there simple Python script that lets you upload an image to Dropbox?

Find more posts tagged with

API

Abuse

Comments

Здравко

Hi @pkd,

I have no idea what you mean "simple Python script", but here is one that handle needed credentials receiving and do some work:

#!/bin/python3
###############################################################################
#        Script uploading files to Dropbox and receiving Dropbox links
#        -------------------------------------------------------------
# Every target file/folder is passed as argument to the script.
# Two preparation steps are needed:
#   1. Register a Dropbox application and put the application key as value of
#      APPLICATION_KEY global variable.
#   2. Register the used REDIRECT_URI in the application just created in
#      previous step. With host 'localhost' and port 8080, the redirect URL
#      that has to be registered is "http://localhost:8080/", for instance.
# Next, just make it executable (if needed), using (on Mac and Linux):
#   $ chmod a+x dropbox_file
# ... and put it in a place visible for execution in your system (somewhere in
# folders pointed by $PATH environment variable). On first run you will be
# invited to link your script to your Dropbox account. When getting links to
# work correct local Dropbox application and this script have to be link to the
# same account!
# For more help type:
#   $ dropbox_file --help
# Author: Здравко
#   www.dropboxforum.com/t5/user/viewprofilepage/user-id/422790
###############################################################################

from dropbox import Dropbox
from dropbox.exceptions import ApiError
from dropbox.files import WriteMode
import json
from pathlib import Path
from datetime import datetime
from os import sep
from sys import exit
import logging
from platformdirs import user_config_path
import click

# Place to save current configuration
CONFIG_JSON=user_config_path('dropbox_file') / 'cred.json'

# Take a look on your application in https://www.dropbox.com/developers/apps
APPLICATION_KEY='PUT YOUR KEY HERE'

URI_HOST='localhost'
URI_PORT=8080
# URI should be registered in the application redirect URIs list!!!
REDIRECT_URI=f"http://{URI_HOST}:{URI_PORT}/"
HOST_PORT=(URI_HOST,URI_PORT)

success_response = (
  "End of authentication flow. <span class="lia-unicode-emoji" title=":winking_face:">😉</span> You're going to do your work!")
cancel_response = "🤷 You have denied your application's work. <span class="lia-unicode-emoji" title=":confused_face:">😕</span>"
error_response = "<span class="lia-unicode-emoji" title=":smiling_face_with_horns:">😈</span> You got an error: "

class ApplicationConfig:
  def __init__(self, conf_path=CONFIG_JSON):
    self.conf_path=conf_path
    self.conf=None
    self.client=None
    self.access_token=None
    self.access_token_expiresat=None
    self.refresh_token=None
    if self.conf_path.is_file():
      try:
        with self.conf_path.open() as fconf:
          self.conf=json.load(fconf)
        self.access_token = self.conf['access_token']
        self.access_token_expiresat = datetime.fromtimestamp(
          self.conf['access_token_expiresat'])
        self.refresh_token = self.conf['refresh_token']
      except Exception:
        self.conf_path.unlink(True)
        self.conf=None
    else:
      conf_path.parent.mkdir(exist_ok=True)
  def __del__(self):
    "Checks for something changed (new access token) and dumps it when there is"
    if (self.client is not None and
        self.client._oauth2_access_token_expiration >
        self.access_token_expiresat):
      self.conf['access_token'] = self.client._oauth2_access_token
      self.conf['access_token_expiresat'] = (
        self.client._oauth2_access_token_expiration.timestamp())
      self.conf['refresh_token'] = self.client._oauth2_refresh_token
      with self.conf_path.open(mode='w') as fconf:
        json.dump(self.conf, fconf)
  def getClient(self):
    "Gets Dropbox client object. Performs OAuth flow if needed."
    if self.conf is None:
      self.client=None
      import webbrowser
      from dropbox import DropboxOAuth2Flow
      from dropbox.oauth import NotApprovedException
      from http.server import HTTPServer, BaseHTTPRequestHandler
      dbxAuth=DropboxOAuth2Flow(APPLICATION_KEY, REDIRECT_URI, {},
        'dropbox-auth-csrf-token', token_access_type='offline', use_pkce=True)
      webbrowser.open(dbxAuth.start())
      conf=None
      conf_path = self.conf_path
      class Handler(BaseHTTPRequestHandler):
        response_success = success_response.encode()
        response_cancel = cancel_response.encode()
        response_error = error_response.encode()
        def do_GET(self):
          nonlocal dbxAuth, conf
          from urllib.parse import urlparse, parse_qs
          query = parse_qs(urlparse(self.path).query)
          for r in query.keys():
            query[r] = query[r][0]
          self.send_response(200)
          self.send_header("content-type", "text/plain;charset=UTF-8")
          try:
            oauthRes = dbxAuth.finish(query)
            conf={'access_token': oauthRes.access_token,
                  'access_token_expiresat': oauthRes.expires_at.timestamp(),
                  'refresh_token': oauthRes.refresh_token}
            with conf_path.open(mode='w') as fconf:
              json.dump(conf, fconf)
          except NotApprovedException:
            conf={}
            self.send_header("content-length",
                             f"{len(Handler.response_cancel)}")
            self.end_headers()
            self.wfile.write(Handler.response_cancel)
            self.wfile.flush()
            return
          except Exception as e:
            conf={}
            r = Handler.response_error + str(e).encode()
            self.send_header("content-length", f"{len(r)}")
            self.end_headers()
            self.wfile.write(r)
            self.wfile.flush()
            return
          self.send_header("content-length",
                           f"{len(Handler.response_success)}")
          self.end_headers()
          self.wfile.write(Handler.response_success)
          self.wfile.flush()
      httpd=HTTPServer((URI_HOST, URI_PORT), Handler)
      while conf is None:
        httpd.handle_request()
      httpd.server_close()
      del httpd
      if 'refresh_token' not in conf:
        raise RuntimeError("Cannot process because missing authentication")
      self.conf = conf
      self.access_token = self.conf['access_token']
      self.access_token_expiresat = datetime.fromtimestamp(
        self.conf['access_token_expiresat'])
      self.refresh_token = self.conf['refresh_token']
    # Makes sure there is cached client object.
    if self.client is None:
      self.client=Dropbox(self.access_token,
                    oauth2_refresh_token=self.refresh_token,
                    oauth2_access_token_expiration=self.access_token_expiresat,
                    app_key=APPLICATION_KEY)
    return self.client

class PathMapper:
  def __init__(self, local=True):
    if not local:
      self.dbx_path = ''
      return
    dbx_info = Path('~/.dropbox/info.json').expanduser()
    if not dbx_info.is_file():
      raise RuntimeError("Missing Dropbox application information")
    with dbx_info.open() as finfo:
      # Only personal accounts are supported by now - group accounts need
      # additional namespace handling (just changing 'personal' is not enough).
      # Somebody else may make some exercises. <span class="lia-unicode-emoji" title=":beaming_face_with_smiling_eyes:">😁</span>
      self.dbx_path = json.load(finfo)['personal']['path']
  def __contains__(self, path):
    path = str(Path(path).expanduser().absolute())
    return ((len(path) == len(self.dbx_path) and path == self.dbx_path) or
            (len(path) > len(self.dbx_path) and path[len(self.dbx_path)] == sep
             and path[:len(self.dbx_path)] == self.dbx_path))
  def __getitem__(self, path):
    path = str(Path(path).expanduser().absolute())
    if ((len(path) == len(self.dbx_path) and path == self.dbx_path) or
        (len(path) > len(self.dbx_path) and path[len(self.dbx_path)] == sep
         and path[:len(self.dbx_path)] == self.dbx_path)):
      return path[len(self.dbx_path):]

pass_config = click.make_pass_decorator(ApplicationConfig, ensure=True)

@click.group()
@click.option('-v', '--verbose', is_flag=True, help="Toggle verbose mode.")
def main(verbose):
  """Perform file actions on files in Dropbox account. Type:
  
  $ dropbox_file COMMAND --help
  
  ... where COMMAND is one of listed below, for command specific options.
  """
  if verbose:
    logging.basicConfig(level=logging.DEBUG)

@main.command()
@click.option('-l', '--local', is_flag=True, help="Try parce path(s) as local "
              "- residing in application's dir (Mac and Linux only).")
@click.argument('paths', type=click.Path(), nargs=-1)
@pass_config
def get_link(conf, paths, local):
  """Takes link(s) to particular file(s) in Dropbox account.
  
  By default paths are passed as rooted to Dropbox account home.
  
  Returns preview URLs (line per file/folder).
  """
  dbxPathMap = PathMapper(local)
  dbx = conf.getClient()
  
  for path in paths:
    if path not in dbxPathMap:
      logging.error(
        f"Passed path '{path}' is not part of the Dropbox driectory tree")
      continue
    dbx_path = dbxPathMap[path]
    if len(dbx_path) == 0:
      logging.error("Dropbox folder itself cannot be pointed by link")
      continue
    logging.debug(f"Processing {'local' if local else 'remote'} file '{path}'"
                  f" with Dropbox path '{dbx_path}'")
    try:
      metadata = dbx.sharing_create_shared_link_with_settings(dbx_path)
    except ApiError as e:
      er = e.error
      if not er.is_shared_link_already_exists():
        raise
      er = er.get_shared_link_already_exists()
      if not er.is_metadata():
        raise
      metadata = er.get_metadata()
    print(metadata.url)

@main.command()
# Can be improved to handle dirs in recursion, but... some other time.
@click.argument('src', type=click.Path(exists=True, dir_okay=False), nargs=-1)
@click.argument('dest', type=click.Path(), nargs=1)
@pass_config
def upload(conf, dest, src):
  """Uploads passed local file(s) to designated place in Dropbox account.
  
  The last argument is the designated place/folder in your Dropbox account.
  This place represent always a target folder. Can be uploaded one or more
  files represented by all arguments but the last one.
  
  Returns associated ID (line per file).
  """
  dbx = conf.getClient()
  dest = Path(dest)
  
  for path in src:
    path = Path(path)
    d = "".join('/'+d for d in dest.joinpath(path.name).parts if d != sep)
    logging.debug(f"Uploading local file '{path}' to Dropbox path '{d}'")
    # Handlig of upload sessions would be nice here for bigger files support
    # and batch upload, but... again some other time.
    with path.open('rb') as f:
      metadata = dbx.files_upload(f.read(), d, WriteMode('overwrite'))
    print(metadata.id)

if __name__ == "__main__":
  try:
    main()
  except Exception as e:
    logging.error(f"Unexpected error: {e}")
    exit(1)

There are lots of things that can be improved, but it can serve as a working template, I think. 😉

Hope this helps.

pkd

Thank you but this is not what I asked for. I thought that there exists simple script that lets you ulpoad files to Dropbox.

I stopped using AWS S3 couple of months ago. I thought Dropbox would be easier in comparison for a small scale cloud storage but apparently it's not.

Thanks anyway.

Здравко

@pkd wrote:
Thank you but this is not what I asked for. I thought that there exists simple script that lets you ulpoad files to Dropbox.
...

🤔Did you test the above script? It can perform upload of one or more files in pointed Dropbox place/folder! The simplest upload command would be:

dropbox_file upload <your local file> ""
dropbox_file upload img.jpg app_folder_dropbox

So, the pointed file would be upload to your home root. If you replace the empty string at the end with some Dropbox folder, then the file will be placed there. Exact syntax can be seen using:

dropbox_file --help

If you mean ready application to sync, yes, there is too. Take a look here.

pkd

Good God, this seems so overly complicated. It's literally 10 lines of code for S3.

I think I'll use some other service instead of Dropbox.

Greg-DB

@pkd I see Здравко helpfully shared a more fully-featured script, but much like what you originally shared, it is possible to upload to Dropbox using the API automatically once you process the authorization flow (which only needs to be done once) with minimal code.

The minimal usage to upload/overwrite a file over extended periods of time would look like this: (using credentials acquired using PKCE, like in Здравко's example)

import dropbox

dbx = dropbox.Dropbox(oauth2_refresh_token="REFRESHTOKENHERE", app_key="APPKEYHERE")

print(dbx.files_upload(b"some data here", "/upload_path/file.ext", mode=dropbox.files.WriteMode("overwrite")))

pkd

I chose "Allow public clients (Implicit Grant & PKCE)" in Settings but still there's no refresh token. Is it even possible to set this whole thing up using web interface at Dropbox?

Здравко

@pkd, Did you get some error or something? 🧐 The refresh token gets in the configuration file (on OAuth flow success) and stay there, where it's read on command start. The file gets updated whenever needed to minimize needed refreshes. Take a look in the configuration file (named 'cert.json', in my script, residing in 'dropbox_file' configuration folder) to take refresh token and use somewhere else if you want to (or copy the entire file). The configuration folder can be found in the standard place for configuration data according your platform. As Greg mentioned the refresh token, once received, can be used in very simplistic scripts without further user authentication. Don't forget to set the needed permissions/scopes, matching to your needs or add the needed as soon as you get error that they're missing (and receive the refresh token anew, such a change is NOT retroactive).

Greg-DB

@pkd Yes, it is possible to use this flow on web. The setting you mention enables flows for client-side applications, but either way you can still request "offline" access to get long-term access via a refresh token, regardless of what kind of platform you're working from.

pkd

@Greg-DBno, I meant is there a way to get refresh token directly on Dropbox website without running some other script?

Здравко

@pkd wrote:
... is there a way to get refresh token directly on Dropbox website without running some other script?

No, there isn't - not yet at least.

PS: Just to mention, if you decide usage of non PKCE OAuth flow, application secret becomes mandatory (in spite optional otherwise)! In such case you would need to add it, as a parameter, on Dropbox client object initialization together with the application key and the refresh token. 😉

Greg-DB

@pkd No, you need to use the OAuth app authorization flow to get a refresh token. Dropbox doesn't offer another way of getting a refresh token.

If you only need this for your own account though, you could run that manually once, e.g., as shown in this example using curl. (The use of redirect_uri is actually optional so you can just omit that parameter.)