Skip to content
Merged
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,4 @@ deps
venv
debug.py
.vscode/
.python-version
33 changes: 27 additions & 6 deletions mergin/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
download_project_cancel,
download_file_async,
download_file_finalize,
download_project_file_async,
download_project_finalize,
download_project_is_running,
)
Expand Down Expand Up @@ -248,16 +249,20 @@ def list_projects(ctx, name, namespace, order_params):
@click.argument("project")
@click.argument("directory", type=click.Path(), required=False)
@click.option("--version", default=None, help="Version of project to download")
@click.option("--include", multiple=True, help="Only download files matching this pattern, e.g. '*.gpkg'")
@click.option("--exclude", multiple=True, help="Skip files matching this pattern, e.g. 'media/*'")
@click.pass_context
def download(ctx, project, directory, version):
def download(ctx, project, directory, version, include, exclude):
"""Download last version of mergin project."""
mc = ctx.obj["client"]
if mc is None:
return
if include and exclude:
raise click.UsageError("--include and --exclude cannot be used together")
directory = directory or os.path.basename(project)
click.echo("Downloading into {}".format(directory))
try:
job = download_project_async(mc, project, directory, version)
job = download_project_async(mc, project, directory, version, include=include, exclude=exclude)
with click.progressbar(length=job.total_size) as bar:
last_transferred_size = 0
while download_project_is_running(job):
Expand Down Expand Up @@ -335,17 +340,33 @@ def share(ctx, project):
@click.argument("filepath")
@click.argument("output")
@click.option("--version", help="Project version tag, for example 'v3'")
@click.option(
"--project",
help="Full project name ('<workspace>/<project>') to download the file directly from the server. "
"If not given, the current directory is used and must be an existing checked out project.",
)
@click.pass_context
def download_file(ctx, filepath, output, version):
def download_file(ctx, filepath, output, version, project):
"""
Download project file at specified version. `project` needs to be a combination of namespace/project.
If no version is given, the latest will be fetched.
Download project file at specified version. If no version is given, the latest will be fetched.
"""
mc = ctx.obj["client"]
if mc is None:
return
try:
job = download_file_async(mc, os.getcwd(), filepath, output, version)
if project is not None:
job = download_project_file_async(mc, project, filepath, output, version)
else:
try:
MerginProject(os.getcwd()).project_full_name()
except InvalidProject:
click.secho(
"Current directory is not a Mergin Maps project. Run this command from within a "
"checked out project directory, or pass --project <workspace>/<project>.",
fg="red",
)
return
job = download_file_async(mc, os.getcwd(), filepath, output, version)
with click.progressbar(length=job.total_size) as bar:
last_transferred_size = 0
while download_project_is_running(job):
Expand Down
36 changes: 34 additions & 2 deletions mergin/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
download_file_async,
download_files_async,
download_files_finalize,
download_project_file_async,
download_diffs_async,
download_project_finalize,
download_project_wait,
Expand All @@ -63,6 +64,7 @@
from .utils import DateTimeEncoder, get_versions_with_file_changes, int_version, is_version_acceptable
from .utils import (
DateTimeEncoder,
filter_files,
get_versions_with_file_changes,
int_version,
is_version_acceptable,
Expand Down Expand Up @@ -902,7 +904,7 @@ def project_versions(self, project_path, since=1, to=None):
filtered_versions = list(filter(lambda v: (num_since <= int_version(v["name"]) <= num_to), versions))
return filtered_versions

def download_project(self, project_path, directory, version=None):
def download_project(self, project_path, directory, version=None, include=None, exclude=None):
"""
Download project into given directory. If version is not specified, latest version is downloaded

Expand All @@ -914,8 +916,16 @@ def download_project(self, project_path, directory, version=None):

:param version: Project version to download, e.g. v42
:type version: String

:param include: Optional list of glob patterns (matched against each file's project path, e.g.
"media/*" or "*.gpkg") - only matching files are downloaded.
:type include: List[String]

:param exclude: Optional list of glob patterns - matching files are skipped. Mutually exclusive
with include.
:type exclude: List[String]
"""
job = download_project_async(self, project_path, directory, version)
job = download_project_async(self, project_path, directory, version, include=include, exclude=exclude)
download_project_wait(job)
download_project_finalize(job)

Expand Down Expand Up @@ -1158,6 +1168,10 @@ def project_status(self, directory):
server_info = self.project_info(mp.project_full_name(), since=mp.version())

pull_changes = mp.get_pull_changes(server_info.get("files", []), server_info.get("version"))
# on a sparse checkout, don't report excluded files as pending server changes -
# they were never meant to be pulled in the first place
file_filter = mp.file_filter()
pull_changes = {change_type: filter_files(files, **file_filter) for change_type, files in pull_changes.items()}

push_changes = mp.get_push_changes()
push_changes_summary = mp.get_list_of_push_changes(push_changes)
Expand Down Expand Up @@ -1212,6 +1226,24 @@ def download_file(self, project_dir, file_path, output_filename, version=None):
pull_project_wait(job)
download_file_finalize(job)

def download_project_file(self, project_path, file_path, output_filename, version=None):
"""
Download a single project file at specified version directly from the server, without
needing an existing local project checkout.

:param project_path: full project name ("<workspace>/<project>")
:type project_path: String
:param file_path: relative path of file to download in the project directory
:type file_path: String
:param output_filename: full destination path for saving the downloaded file
:type output_filename: String
:param version: optional version tag for downloaded file
:type version: String
"""
job = download_project_file_async(self, project_path, file_path, output_filename, version=version)
pull_project_wait(job)
download_file_finalize(job)

def get_file_diff(self, project_dir, file_path, output_diff, version_from, version_to):
"""Create concatenated diff for project file diffs between versions version_from and version_to.

Expand Down
96 changes: 80 additions & 16 deletions mergin/client_pull.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,8 @@
from .common import CHUNK_SIZE, ClientError, DeltaChangeType, PullActionType
from .models import ProjectDelta, ProjectDeltaChange, PullAction
from .merginproject import MerginProject
from .utils import cleanup_tmp_dir, save_to_file
from typing import List, Optional
from .utils import cleanup_tmp_dir, filter_files, is_path_in_scope, save_to_file, is_versioned_file
from typing import List, Optional, Union

# status = download_project_async(...)
#
Expand Down Expand Up @@ -54,7 +54,7 @@ def __init__(
update_tasks,
download_queue_items,
tmp_dir: tempfile.TemporaryDirectory,
mp,
mp: Union[MerginProject, "DownloadScratchContext"],
project_info,
):
self.project_path = project_path
Expand All @@ -64,7 +64,7 @@ def __init__(
self.update_tasks = update_tasks
self.download_queue_items = download_queue_items
self.tmp_dir = tmp_dir
self.mp = mp # MerginProject instance
self.mp = mp
self.is_cancelled = False
self.project_info = project_info # parsed JSON with project info returned from the server
self.failure_log_file = None # log file, copied from the project directory if download fails
Expand All @@ -80,6 +80,25 @@ def dump(self):
print("--- END ---")


class DownloadScratchContext:
"""
Minimal stand-in for MerginProject used when downloading a file directly by
project name ("<workspace>/<project>") without an existing local project checkout.

Provides only what the shared download job code actually needs from MerginProject.
"""

def __init__(self, mc, cache_dir: str):
self.log = mc.log
self.cache_dir = cache_dir
# only used by _cleanup_failed_download() to look for a log file
self.dir = cache_dir

def remove_logging_handler(self):
# no-op: self.log is mc's shared logger, not owned by this throwaway context
pass


class DownloadQueueItem:
"""
a piece of data from a project that should be downloaded - it can be either a chunk or it can be a diff.
Expand Down Expand Up @@ -242,12 +261,18 @@ def _cleanup_failed_download(mergin_project: MerginProject = None):
return dest_path


def download_project_async(mc, project_path, directory, project_version=None):
def download_project_async(mc, project_path, directory, project_version=None, include=None, exclude=None):
"""
Starts project download in background and returns handle to the pending project download.
Using that object it is possible to watch progress or cancel the ongoing work.

`include`/`exclude` are optional lists of glob patterns (matched against each file's project
path, e.g. "media/*" or "*.gpkg") to only download a subset of the project's files. They are
mutually exclusive.
"""

if include and exclude:
raise ClientError("Cannot use both include and exclude filters at the same time")
if "/" not in project_path:
raise ClientError("Project name needs to be fully qualified, e.g. <username>/<projectname>")
if os.path.exists(directory):
Expand Down Expand Up @@ -276,6 +301,12 @@ def download_project_async(mc, project_path, directory, project_version=None):

mp.log.info(f"got project info. version {version}")

# keep only the files matching the filter (if any)
project_info["files"] = filter_files(project_info["files"], include=include, exclude=exclude)
# persisted once since it must never change again for this checkout
if include or exclude:
mp.write_file_filter({"include": include, "exclude": exclude})

# prepare download
update_tasks = [] # stuff to do at the end of download
for file in project_info["files"]:
Expand Down Expand Up @@ -398,7 +429,7 @@ def __init__(
self.download_queue_items = download_queue_items
self.latest_version = latest_version

def apply(self, directory, mp):
def apply(self, directory, mp: Union[MerginProject, "DownloadScratchContext"]):
"""assemble downloaded chunks into a single file"""

if self.destination_file is None:
Expand All @@ -411,14 +442,14 @@ def apply(self, directory, mp):
os.makedirs(file_dir, exist_ok=True)

# ignore check if we download not-latest version of gpkg file (possibly reconstructed on server on demand)
check_size = self.latest_version or not mp.is_versioned_file(self.file_path)
check_size = self.latest_version or not is_versioned_file(self.file_path)
# merge chunks together (and delete them afterwards)
file_to_merge = DownloadFile(dest_file_path, self.download_queue_items, check_size)
file_to_merge.from_chunks()

# Make a copy of the file to meta dir only if there is no user-specified path for the file.
# destination_file is None for full project download and takes a meaningful value for a single file download.
if mp.is_versioned_file(self.file_path) and self.destination_file is None:
# destination_file is None for full project download and takes a meaningful value for a single file download
if self.destination_file is None and is_versioned_file(self.file_path):
mp.geodiff.make_copy_sqlite(mp.fpath(self.file_path), mp.fpath_meta(self.file_path))


Expand Down Expand Up @@ -525,6 +556,9 @@ def pull_project_async(mc, directory) -> Optional[PullJob]:
mp.log.info("--- pull aborted")
raise

file_filter = mp.file_filter()
delta.changes = [c for c in delta.changes if is_path_in_scope(c.path, **file_filter)]

mp.log.info(f"got project versions: local version {local_version} / server version {server_version}")

if local_version == server_version:
Expand Down Expand Up @@ -748,6 +782,9 @@ def pull_project_finalize(job: PullJob):
cleanup_tmp_dir(job.mp, job.tmp_dir) # delete our temporary dir and all its content
raise ClientError("Failed to apply pull actions: " + str(e))

# keep only in-scope files in the metadata we're about to persist
job.project_info["files"] = filter_files(job.project_info["files"], **job.mp.file_filter())

job.mp.update_metadata(job.project_info)

if job.mp.has_unfinished_pull():
Expand All @@ -774,6 +811,23 @@ def download_file_finalize(job):
download_files_finalize(job)


def download_project_file_async(mc, project_path: str, file_path: str, output_file: str, version: str = None):
"""
Starts background download of a single project file at specified version, fetched directly
from the server without needing an existing local project checkout.
Returns handle to the pending download.

:param project_path: full project name ("<workspace>/<project>")
:param output_file: destination path for the downloaded file
"""
if not output_file:
raise ClientError("output_file must be provided when downloading a file without a local project checkout")

tmp_dir = tempfile.TemporaryDirectory(prefix="python-api-client-")
mp = DownloadScratchContext(mc, tmp_dir.name)
return _download_files_async(mc, mp, project_path, [file_path], [output_file], version, tmp_dir)


def download_diffs_async(mc, project_directory, file_path, versions):
"""
Starts background download project file diffs for specified versions.
Expand Down Expand Up @@ -897,14 +951,29 @@ def download_diffs_finalize(job: PullJob) -> List[str]:


def download_files_async(
mc, project_dir: str, file_paths: typing.List[str], output_paths: typing.List[str], version: str
mc, project_dir: str, file_paths: typing.List[str], output_paths: typing.List[str] = None, version: str = None
):
"""
Starts background download project files at specified version.
Returns handle to the pending download.

`project_dir` must be an existing local project directory.
"""
mp = MerginProject(project_dir)
project_path = mp.project_full_name()
tmp_dir = tempfile.TemporaryDirectory(prefix="python-api-client-")
return _download_files_async(mc, mp, project_path, file_paths, output_paths, version, tmp_dir)


def _download_files_async(
mc,
mp: Union[MerginProject, "DownloadScratchContext"],
project_path: str,
file_paths: typing.List[str],
output_paths: typing.List[str],
version: str,
tmp_dir: tempfile.TemporaryDirectory,
):
ver_info = f"at version {version}" if version is not None else "at latest version"
mp.log.info(f"Getting [{', '.join(file_paths)}] {ver_info}")
latest_proj_info = mc.project_info(project_path)
Expand All @@ -914,9 +983,6 @@ def download_files_async(
project_info = latest_proj_info
mp.log.info(f"Got project info. version {project_info['version']}")

# set temporary directory for download
tmp_dir = tempfile.TemporaryDirectory(prefix="python-api-client-")

if output_paths is None:
output_paths = []
for file in file_paths:
Expand Down Expand Up @@ -991,6 +1057,4 @@ def download_files_finalize(job: DownloadJob):
for task in job.update_tasks:
task.apply(job.tmp_dir, job.mp)

# Remove temporary download directory
if job.tmp_dir is not None and os.path.exists(job.tmp_dir.name):
cleanup_tmp_dir(job.mp, job.tmp_dir)
cleanup_tmp_dir(job.mp, job.tmp_dir)
5 changes: 4 additions & 1 deletion mergin/client_push.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@
)
from .merginproject import MerginProject, pygeodiff
from .editor import filter_changes
from .utils import get_data_checksum, cleanup_tmp_dir
from .utils import get_data_checksum, cleanup_tmp_dir, filter_files

POST_JSON_HEADERS = {"Content-Type": "application/json"}

Expand Down Expand Up @@ -458,6 +458,9 @@ def push_project_finalize(job: UploadJob):
cleanup_tmp_dir(job.mp, job.tmp_dir) # delete our temporary dir and all its content
raise err

# keep only in-scope files in the metadata we're about to persist
job.server_resp["files"] = filter_files(job.server_resp["files"], **job.mp.file_filter())

job.mp.update_metadata(job.server_resp)
try:
job.mp.apply_push_changes(asdict(job.changes))
Expand Down
Loading
Loading