DEV Community

Cover image for Creating a real-time local to cloud synchronizer in Python
Samuel Viana
Samuel Viana

Posted on • Originally published at codehouse.digfish.org

Creating a real-time local to cloud synchronizer in Python

The idea

Some cloud storage providers offer a client application that can be installed in the local storage, allowing for a "local mirror" of the remote storage. Usually this synchronization is two-way, the changes on the local system are propagated to the cloud storage and the changes made in the remote storage are reflected in the local mirror. Dropbox since its inception as such app for such purpose, Google Drive used to have one of these, and OneDrive also have such application. In this article, I will present a 'proof-of-concept' Python application that is able to make the "first part" of the synchronize, ie, reproduce the local changes in the remote cloud storage.

The implementation

Using the official Dropbox Python SDK and the library watchdog, it's possible to crate one such application in around 100 lines of code. Watchdog allows you to monitor changes in your local file system through events triggered by those changes like 'FileCreated','FileModified','FileDeleted' or 'FileMoved'. For each type of event, there one respective method in an EventHandler subclass which should implement these methods. To use the Dropbox SDK, you'll have to create one application in the Developer frontend. Generate one access token that can be used in our application. This application just have two parameters: the local directory where the file changes will occur and the remote directory on Dropbox where these changes should be reflect. It's a kind of mapping between that local directory and this remote directory.

NOTE: This code does not perform an initial synchronization of the local files to the directory in the cloud. It only uploads file created or changed during the execution of the program.

# directory sychornizer using python watchdog library with dropbox api
import os, os.path
import logging
import dropbox
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, LoggingEventHandler
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
root_dropbox_path = '/tmp'
local_path = 'D:/temp/dropbox-tmp'
class LocalHandler(FileSystemEventHandler):
def __init__(self,dbx):
#super().__init__()
self.dbx = dbx
def on_created(self, evt):
logging.info(evt)
if evt.is_directory:
dir_to_create = root_dropbox_path + '/' + os.path.relpath(evt.src_path,local_path)
logging.info(f"Created directory on Dropbox:{dir_to_create}")
try:
self.dbx.files_create_folder(dir_to_create)
except dropbox.exceptions.ApiError as err:
logging.error(err)
def on_modified(self,evt):
logging.info(evt)
if not evt.is_directory:
f = open(evt.src_path,'rb')
path_to_upload = os.path.relpath(evt.src_path,local_path)
dbx_path_target = root_dropbox_path + '/' + path_to_upload
logging.info(f"{path_to_upload} => Dropbox:{dbx_path_target}")
try:
self.dbx.files_upload(f.read(),dbx_path_target )
except dropbox.exceptions.ApiError as err:
logging.error(err)
def on_move(self,evt):
logging.info(evt)
dropbox_src_path = root_dropbox_path + '/' + os.path.relpath(evt.src_path,local_path)
dropbox_dest_path = root_dropbox_path + '/' + os.path.relpath(evt.dest_path,local_path)
logging.info(f"Move on Dropbox:{dropbox_src_path} => {dropbox_dest_path}")
try:
self.dbx.files_move(dropbox_src_path,dropbox_dest_path)
except dropbox.exceptions.ApiError as err:
logging.error(err)
def on_deleted(self, event):
logging.info(event)
path_to_delete = root_dropbox_path + '/'+ os.path.relpath(event.src_path,local_path)
logging.info(f"Delete on Dropbox:{path_to_delete}")
try:
self.dbx.files_to_delete(path_to_delete)
except dropbox.exceptions.ApiError as err:
logging.error(err)
def main():
dbx = dropbox.Dropbox("YOUR ACCESS TOKEN HERE")
logging.info(dbx.users_get_current_account())
event_handler = LocalHandler(dbx)
#event_handler = LoggingEventHandler()
observer = Observer()
observer.schedule(event_handler, local_path, recursive=True)
observer.start()
try:
while observer.is_alive():
observer.join(1)
finally:
observer.stop()
observer.join()
if __name__ == "__main__":
main()

Heroku

Build apps, not infrastructure.

Dealing with servers, hardware, and infrastructure can take up your valuable time. Discover the benefits of Heroku, the PaaS of choice for developers since 2007.

Visit Site

Top comments (0)

Retry later