Merge pull request #8 from lutraconsulting/upload_revisited

Upload revisited

Author: sklencar

CHANGELOG.md

@@ -4,8 +4,8 @@
 
 - Added filters for listing projects (owner, shared, query)
 - Changed Basic auth to Bearer token-based auth
-- Improved CLI: added login, credentials in environment variables
-- Fixed missing Content-Length header in upload request
+- Improved CLI: added login, credentials in env variables, delete project
+- Download/upload files with multiple sequential requests (chunked transfer)
 
 ## 2019.3
 

README.md

@@ -1,9 +1,16 @@
 # Mergin python client
 
-Repo for mergin client and basic utils.
+Repo for [mergin](https://public.cloudmergin.com/) client and basic utils.
+
+Python 3.0+ required.
 
 For using mergin client with its dependencies locally run:
 
+    pip install wheel
     python3 setup.py sdist bdist_wheel
     mkdir -p mergin/deps
-    pip wheel -r mergin_client.egg-info/requires.txt -w mergin/deps
+    pip wheel -r mergin_client.egg-info/requires.txt -w mergin/deps
+
+To run cli you need to install click:
+
+    pip install click

cli.py

@@ -51,13 +51,14 @@ def pretty_diff(diff):
 def _init_client():
     url = os.environ.get('MERGIN_URL')
     auth_token = os.environ.get('MERGIN_AUTH')
-    return MerginClient(url, auth_token=auth_token)
+    return MerginClient(url, auth_token='Bearer {}'.format(auth_token))
 
 
 @click.group()
 def cli():
     pass
 
+
 @cli.command()
 @click.argument('url')
 @click.option('--login', prompt=True)
@@ -67,7 +68,9 @@ def login(url, login, password):
     c = MerginClient(url)
     session = c.login(login, password)
     print('export MERGIN_URL="%s"' % url)
-    print('export MERGIN_AUTH="Bearer %s"' % session['token'])
+    print('export MERGIN_AUTH="%s"' % session['token'])
+    print('export MERGIN_AUTH_HEADER="Authorization: %s"' % session['token'])
+
 
 @cli.command()
 @click.argument('directory', type=click.Path(exists=True))
@@ -85,6 +88,7 @@ def init(directory, public):
     except Exception as e:
         click.secho(str(e), fg='red')
 
+
 @cli.command()
 @click.argument('project')
 @click.argument('directory', type=click.Path(), required=False)
@@ -155,6 +159,7 @@ def status():
         click.secho("No local changes!", fg="magenta")
     # TODO: show conflicts
 
+
 @cli.command()
 def push():
     """Upload local changes into Mergin repository"""
@@ -168,6 +173,7 @@ def push():
     except Exception as e:
         click.secho(str(e), fg='red')
 
+
 @cli.command()
 def pull():
     """Fetch changes from Mergin repository"""
@@ -179,6 +185,7 @@ def pull():
     except InvalidProject:
         click.secho('Invalid project directory', fg='red')
 
+
 @cli.command()
 def version():
     """Check and display server version"""
@@ -189,6 +196,7 @@ def version():
     if not ok:
         click.secho("Server doesn't meet the minimum required version: %s" % c.min_server_version, fg='yellow')
 
+
 @cli.command()
 @click.argument('directory', required=False)
 def modtime(directory):
@@ -209,5 +217,30 @@ def modtime(directory):
             click.echo()
 
 
+@cli.command()
+@click.argument('project', required=False)
+def remove(project):
+    """Remove project from server and locally (if exists)."""
+    local_info = None
+    if not project:
+        from mergin.client import inspect_project
+        try:
+            local_info = inspect_project(os.path.join(os.getcwd()))
+            project = local_info['name']
+        except InvalidProject:
+            click.secho('Invalid project directory', fg='red')
+            return
+
+    c = _init_client()
+    try:
+        c.delete_project(project)
+        if local_info:
+            import shutil
+            shutil.rmtree(os.path.join(os.getcwd()))
+        click.echo('Done')
+    except Exception as e:
+        click.secho(str(e), fg='red')
+
+
 if __name__ == '__main__':
     cli()

mergin/client.py

@@ -6,15 +6,17 @@
 import shutil
 import urllib.parse
 import urllib.request
-from datetime import datetime
+import uuid
+import math
+import hashlib
+from datetime import datetime, timezone
 
 this_dir = os.path.dirname(os.path.realpath(__file__))
 CHUNK_SIZE = 10 * 1024 * 1024
 
 try:
-    from requests_toolbelt import MultipartEncoder
-    import pytz
     import dateutil.parser
+    from dateutil.tz import tzlocal
 except ImportError:
     # this is to import all dependencies shipped with package (e.g. to use in qgis-plugin)
     deps_dir = os.path.join(this_dir, 'deps')
@@ -23,11 +25,10 @@
         for f in os.listdir(os.path.join(deps_dir)):
             sys.path.append(os.path.join(deps_dir, f))
 
-        from requests_toolbelt import MultipartEncoder
-        import pytz
         import dateutil.parser
+        from dateutil.tz import tzlocal
 
-from .utils import generate_checksum, move_file, save_to_file
+from .utils import save_to_file, generate_checksum, move_file, DateTimeEncoder
 
 
 class InvalidProject(Exception):
@@ -54,11 +55,14 @@ def list_project_directory(directory):
         dirs[:] = [d for d in dirs if d not in excluded_dirs]
         for file in files:
             abs_path = os.path.abspath(os.path.join(root, file))
-            proj_path = abs_path[len(prefix) + 1:]
+            rel_path = os.path.relpath(abs_path, start=prefix)
+            # we need posix path
+            proj_path = '/'.join(rel_path.split(os.path.sep))
             proj_files.append({
                 "path": proj_path,
                 "checksum": generate_checksum(abs_path),
-                "size": os.path.getsize(abs_path)
+                "size": os.path.getsize(abs_path),
+                "mtime": datetime.fromtimestamp(os.path.getmtime(abs_path), tzlocal())
             })
     return proj_files
 
@@ -158,7 +162,7 @@ def __init__(self, url, auth_token=None, login=None, password=None):
 
     def _do_request(self, request):
         if self._auth_session:
-            delta = self._auth_session["expire"] - datetime.now(pytz.utc)
+            delta = self._auth_session["expire"] - datetime.now(timezone.utc)
             if delta.total_seconds() < 1:
                 self._auth_session = None
                 # Refresh auth token when login credentials are available
@@ -179,17 +183,14 @@ def get(self, path, data=None, headers={}):
         url = urllib.parse.urljoin(self.url, urllib.parse.quote(path))
         if data:
             url += "?" + urllib.parse.urlencode(data)
-        if headers:
-            request = urllib.request.Request(url, headers=headers)
-        else:
-            request = urllib.request.Request(url)
+        request = urllib.request.Request(url, headers=headers)
         return self._do_request(request)
 
-    def post(self, path, data, headers={}):
+    def post(self, path, data=None, headers={}):
         url = urllib.parse.urljoin(self.url, urllib.parse.quote(path))
         if headers.get("Content-Type", None) == "application/json":
-            data = json.dumps(data).encode("utf-8")
-        request = urllib.request.Request(url, data, headers)
+            data = json.dumps(data, cls=DateTimeEncoder).encode("utf-8")
+        request = urllib.request.Request(url, data, headers, method="POST")
         return self._do_request(request)
 
     def server_version(self):
@@ -247,7 +248,7 @@ def login(self, login, password):
         data = json.load(resp)
         session = data["session"]
         self._auth_session = {
-            "token" : session["token"],
+            "token": "Bearer %s" % session["token"],
             "expire": dateutil.parser.parse(session["expire"])
         }
         self._user_info = {
@@ -281,11 +282,12 @@ def create_project(self, project_name, directory, is_public=False):
         self.post("/v1/project/%s" % namespace, params, {"Content-Type": "application/json"})
         data = {
             "name": "%s/%s" % (namespace, project_name),
-            "version": "",
-            "files": None
+            "version": "v0",
+            "files": []
         }
         save_project_file(directory, data)
-        self.push_project(directory)
+        if len(os.listdir(directory)) > 1:
+            self.push_project(directory)
 
     def projects_list(self, tags=None, user=None, flag=None, q=None):
         """
@@ -379,37 +381,49 @@ def push_project(self, directory):
         local_info = inspect_project(directory)
         project_path = local_info["name"]
         server_info = self.project_info(project_path)
-        if local_info.get("version", "") != server_info.get("version", ""):
-            raise Exception("Update your local repository")
+        server_version = server_info["version"] if server_info["version"] else "v0"
+        if local_info.get("version", "v0") != server_version:
+            raise ClientError("Update your local repository")
 
         files = list_project_directory(directory)
         changes = project_changes(server_info["files"], files)
-        count = sum(len(items) for items in changes.values())
-        if count:
-            # Custom MultipartEncoder doesn't compute Content-Length,
-            # which is currently required by gunicorn.
-            # def fields():
-            #     yield Field("changes", json.dumps(changes).encode("utf-8"))
-            #     for file in (changes["added"] + changes["updated"]):
-            #         path = file["path"]
-            #         with open(os.path.join(directory, path), "rb") as f:
-            #             yield Field(path, f, filename=path, content_type="application/octet-stream")
-            # encoder = MultipartEncoder(fields())
-
-            fields = {"changes": json.dumps(changes).encode("utf-8")}
-            for file in (changes["added"] + changes["updated"]):
-                path = file["path"]
-                fields[path] = (path, open(os.path.join(directory, path), 'rb'), "application/octet-stream")
-            encoder = MultipartEncoder(fields=fields)
-            headers = {
-                "Content-Type": encoder.content_type,
-                "Content-Length": encoder.len
-            }
-            resp = self.post("/v1/project/data_sync/{}".format(project_path), encoder, headers=headers)
-            new_project_info = json.load(resp)
-            local_info["files"] = new_project_info["files"]
-            local_info["version"] = new_project_info["version"]
-            save_project_file(directory, local_info)
+        upload_files = changes["added"] + changes["updated"]
+
+        for f in upload_files:
+            f["chunks"] = [str(uuid.uuid4()) for i in range(math.ceil(f["size"] / CHUNK_SIZE))]
+
+        data = {
+            "version": local_info.get("version"),
+            "changes": changes
+        }
+        resp = self.post("/v1/project/push/%s" % project_path, data, {"Content-Type": "application/json"})
+        info = json.load(resp)
+
+        # upload files' chunks and close transaction
+        if upload_files:
+            headers = {"Content-Type": "application/octet-stream"}
+            for f in upload_files:
+                with open(os.path.join(directory, f["path"]), 'rb') as file:
+                    for chunk in f["chunks"]:
+                        data = file.read(CHUNK_SIZE)
+                        checksum = hashlib.sha1()
+                        checksum.update(data)
+                        size = len(data)
+                        resp = self.post("/v1/project/push/chunk/%s/%s" % (info["transaction"], chunk), data, headers)
+                        data = json.load(resp)
+                        if not (data['size'] == size and data['checksum'] == checksum.hexdigest()):
+                            self.post("/v1/project/push/cancel/%s" % info["transaction"])
+                            raise ClientError("Mismatch between uploaded file and local one")
+            try:
+                resp = self.post("/v1/project/push/finish/%s" % info["transaction"])
+                info = json.load(resp)
+            except ClientError:
+                self.post("/v1/project/push/cancel/%s" % info["transaction"])
+                raise
+
+        local_info["files"] = info["files"]
+        local_info["version"] = info["version"]
+        save_project_file(directory, local_info)
 
     def pull_project(self, directory):
         """
@@ -447,6 +461,7 @@ def backup_if_conflict(path, checksum):
                     while os.path.exists(backup_path):
                         backup_path = local_path("{}_conflict_copy{}".format(path, index))
                         index += 1
+                    # it is unnecessary to copy conflicted file, it would be better to simply rename it
                     shutil.copy(local_path(path), backup_path)
 
         fetch_files = pull_changes["added"] + pull_changes["updated"]
@@ -507,3 +522,16 @@ def _download_file(self, project_path, project_version, file, directory):
                 with open(os.path.join(directory, file['path'] + ".{}".format(i)), 'rb') as chunk:
                     shutil.copyfileobj(chunk, final)
                 os.remove(os.path.join(directory, file['path'] + ".{}".format(i)))
+
+    def delete_project(self, project_path):
+        """
+        Delete project repository on server.
+
+        :param project_path: Project's full name (<namespace>/<name>)
+        :type project_path: String
+
+        """
+        path = "/v1/project/%s" % project_path
+        url = urllib.parse.urljoin(self.url, urllib.parse.quote(path))
+        request = urllib.request.Request(url, method="DELETE")
+        self._do_request(request)

mergin/utils.py

@@ -1,6 +1,8 @@
 import os
 import io
+import json
 import hashlib
+from datetime import datetime
 
 
 def generate_checksum(file, chunk_size=4096):
@@ -45,3 +47,11 @@ def move_file(src, dest):
     if not os.path.exists(dest_dir):
         os.makedirs(dest_dir)
     os.rename(src, dest)
+
+
+class DateTimeEncoder(json.JSONEncoder):
+    def default(self, obj):
+        if isinstance(obj, datetime):
+            return obj.isoformat()
+
+        return super().default(obj)

setup.py

@@ -17,9 +17,7 @@
 
     platforms='any',
     install_requires=[
-        'pytz',
-        'python-dateutil',
-        'requests_toolbelt'
+        'python-dateutil'
     ],
 
     test_suite='nose.collector',