API documentation

DEFAULT_PAGINATION_LIMIT = 20 module-attribute

int: Defines the default limit for pagination, set to 20.

Client

The core component of the QFieldCloud SDK, providing methods for interacting with the QFieldCloud platform.

The client provides methods for authentication, project management, file management, and more.

It is configured with retries for GET requests on specific 5xx HTTP status codes.

Parameters:

Name	Type	Description	Default
url	str	The base URL for the QFieldCloud API. Defaults to QFIELDCLOUD_URL environment variable if empty.	''
verify_ssl	bool	Whether to verify SSL certificates. Defaults to True.	True
token	str	The authentication token for API access. Defaults to QFIELDCLOUD_TOKEN environment variable if empty.	''

Raises:

Type	Description
QfcException	If the url is not provided either directly or through the environment variable.

Attributes:

Name	Type	Description
session	Session	The session object to maintain connections.
url	str	The base URL for the QFieldCloud API.
token	str	The authentication token for API access.
verify_ssl	bool	Whether to verify SSL certificates.

Source code in qfieldcloud_sdk/sdk.py

class Client:
    """The core component of the QFieldCloud SDK, providing methods for interacting with the QFieldCloud platform.

    The client provides methods for authentication, project management, file management, and more.

    It is configured with retries for GET requests on specific 5xx HTTP status codes.

    Args:
        url: The base URL for the QFieldCloud API. Defaults to `QFIELDCLOUD_URL` environment variable if empty.
        verify_ssl: Whether to verify SSL certificates. Defaults to True.
        token: The authentication token for API access. Defaults to `QFIELDCLOUD_TOKEN` environment variable if empty.

    Raises:
        QfcException: If the `url` is not provided either directly or through the environment variable.

    Attributes:
        session: The session object to maintain connections.
        url: The base URL for the QFieldCloud API.
        token: The authentication token for API access.
        verify_ssl: Whether to verify SSL certificates.
    """

    session: requests.Session

    url: str

    token: str

    verify_ssl: bool

    def __init__(
        self,
        url: str = "",
        verify_ssl: bool = True,
        token: str = "",
    ) -> None:
        self.session = requests.Session()
        # retries should be only on GET and only if error 5xx
        retries = Retry(
            total=5,
            backoff_factor=0.1,
            allowed_methods=["GET"],
            # skip 501, as it is "Not Implemented", no point to retry
            status_forcelist=[500, 502, 503, 504],
        )
        self.session.mount("https://", HTTPAdapter(max_retries=retries))

        self.url = url or os.environ.get("QFIELDCLOUD_URL", "")
        self.token = token or os.environ.get("QFIELDCLOUD_TOKEN", "")
        self.verify_ssl = verify_ssl

        if not self.verify_ssl:
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

        if not self.url:
            raise QfcException(
                "Cannot create a new QFieldCloud client without a url passed in the constructor or as environment variable QFIELDCLOUD_URL"
            )

    def login(self, username: str, password: str) -> Dict[str, Any]:
        """Logs in with the provided username and password.

        Args:
            username: The username or email used to register.
            password: The password associated with the username.

        Returns:
            Authentication token and additional metadata.

        Example:
            client = sdk.Client(url="https://app.qfield.cloud/api/v1/")
            client.login("ninjamaster", "secret_password123")
        """
        resp = self._request(
            "POST",
            "auth/login",
            data={
                "username": username,
                "password": password,
            },
            skip_token=True,
        )

        payload = resp.json()

        self.token = payload["token"]

        return payload

    def logout(self) -> None:
        """Logs out from the current session, invalidating the authentication token.

        Example:
            client.logout()
        """
        resp = self._request("POST", "auth/logout")

        return resp.json()

    def list_projects(
        self,
        include_public: bool = False,
        pagination: Pagination = Pagination(),
        **kwargs,
    ) -> List[Dict[str, Any]]:
        """List projects accessible by the current user. Optionally include all public projects.

        Args:
            include_public: Whether to include public projects in the list. Defaults to False.
            pagination: Pagination settings for the request. Defaults to an empty Pagination instance.

        Returns:
            A list of dictionaries containing project details.
        """
        params = {
            "include-public": str(int(include_public)),  # type: ignore
        }

        payload = self._request_json(
            "GET", "projects", params=params, pagination=pagination
        )
        return cast(List, payload)

    def list_remote_files(
        self, project_id: str, skip_metadata: bool = True
    ) -> List[Dict[str, Any]]:
        """List project files.

        Args:
            project_id: Project ID.
            skip_metadata: Whether to skip fetching metadata for the files. Defaults to True.

        Returns:
            A list of file details.

        Example:
            client.list_remote_files("project_id", True)
        """
        params = {}

        if skip_metadata:
            params["skip_metadata"] = "1"

        resp = self._request("GET", f"files/{project_id}", params=params)
        remote_files = resp.json()
        # TODO remove this temporary decoration with `etag` key
        remote_files = list(map(lambda f: {"etag": f["md5sum"], **f}, remote_files))

        return remote_files

    def create_project(
        self,
        name: str,
        owner: Optional[str] = None,
        description: str = "",
        is_public: bool = False,
    ) -> Dict[str, Any]:
        """Create a new project.

        Args:
            name: The name of the new project.
            owner: The owner of the project. When None, the project will be owned by the currently logged-in user. Defaults to None.
            description: A description of the project. Defaults to an empty string.
            is_public: Whether the project should be public. Defaults to False.

        Returns:
            A dictionary containing the details of the created project.
        """
        resp = self._request(
            "POST",
            "projects",
            data={
                "name": name,
                "owner": owner,
                "description": description,
                "is_public": int(is_public),
            },
        )

        return resp.json()

    def delete_project(self, project_id: str) -> requests.Response:
        """Delete a project.

        Args:
            project_id: Project ID.

        Returns:
            The response object from the file delete request.
        """
        resp = self._request("DELETE", f"projects/{project_id}")

        return resp

    def upload_files(
        self,
        project_id: str,
        upload_type: FileTransferType,
        project_path: str,
        filter_glob: str,
        throw_on_error: bool = False,
        show_progress: bool = False,
        force: bool = False,
        job_id: str = "",
    ) -> List[Dict]:
        """Upload files to a QFieldCloud project.

        Args:
            project_id: Project ID.
            upload_type: The type of file transfer (PROJECT or PACKAGE).
            project_path: The local directory containing the files to upload.
            filter_glob: A glob pattern to filter which files to upload.
            throw_on_error: Whether to raise an error if a file fails to upload. Defaults to False.
            show_progress: Whether to display a progress bar during upload. Defaults to False.
            force: Whether to force upload all files, even if they exist remotely. Defaults to False.
            job_id: The job ID, required if `upload_type` is PACKAGE. Defaults to an empty string.

        Returns:
            A list of dictionaries with information about the uploaded files.
        """
        if not filter_glob:
            filter_glob = "*"

        local_files = self.list_local_files(project_path, filter_glob)

        # we should always upload all package files
        if upload_type == FileTransferType.PACKAGE:
            force = True

        files_to_upload = []
        if force:
            files_to_upload = local_files
        else:
            remote_files = self.list_remote_files(project_id)

            if len(remote_files) == 0:
                files_to_upload = local_files
            else:
                for local_file in local_files:
                    remote_file = None
                    for f in remote_files:
                        if f["name"] == local_file["name"]:
                            remote_file = f
                            break

                    etag = calc_etag(local_file["absolute_filename"])
                    if remote_file and remote_file.get("etag", None) == etag:
                        continue

                    files_to_upload.append(local_file)

        if not files_to_upload:
            return files_to_upload

        for file in files_to_upload:
            try:
                local_filename = Path(file["absolute_filename"])
                self.upload_file(
                    project_id,
                    upload_type,
                    local_filename,
                    file["name"],
                    show_progress,
                    job_id,
                )
                file["status"] = FileTransferStatus.SUCCESS
            except Exception as err:
                file["status"] = FileTransferStatus.FAILED
                file["error"] = err

                if throw_on_error:
                    raise err
                else:
                    continue

        return local_files

    def upload_file(
        self,
        project_id: str,
        upload_type: FileTransferType,
        local_filename: Path,
        remote_filename: Path,
        show_progress: bool,
        job_id: str = "",
    ) -> requests.Response:
        """Upload a single file to a project.

        Args:
            project_id: Project ID.
            upload_type: The type of file transfer.
            local_filename: The path to the local file to upload.
            remote_filename: The path where the file should be stored remotely.
            show_progress: Whether to display a progress bar during upload.
            job_id: The job ID, required if `upload_type` is PACKAGE. Defaults to an empty string.

        Returns:
            The response object from the upload request.
        """
        with open(local_filename, "rb") as local_file:
            upload_file = local_file
            if show_progress:
                from tqdm import tqdm
                from tqdm.utils import CallbackIOWrapper

                progress_bar = tqdm(
                    total=local_filename.stat().st_size,
                    unit_scale=True,
                    desc=local_filename.stem,
                )
                upload_file = CallbackIOWrapper(progress_bar.update, local_file, "read")
            else:
                logger.info(f'Uploading file "{remote_filename}"…')

            if upload_type == FileTransferType.PROJECT:
                url = f"files/{project_id}/{remote_filename}"
            elif upload_type == FileTransferType.PACKAGE:
                if not job_id:
                    raise QfcException(
                        'When the upload type is "package", you must pass the "job_id" parameter.'
                    )

                url = f"packages/{project_id}/{job_id}/files/{remote_filename}"
            else:
                raise NotImplementedError()

            return self._request(
                "POST",
                url,
                files={
                    "file": upload_file,
                },
            )

    def download_project(
        self,
        project_id: str,
        local_dir: str,
        filter_glob: str = "*",
        throw_on_error: bool = False,
        show_progress: bool = False,
        force_download: bool = False,
    ) -> List[Dict]:
        """Download the specified project files into the destination dir.

        Args:
            project_id: Project ID.
            local_dir: destination directory where the files will be downloaded
            filter_glob: if specified, download only project files which match the glob, otherwise download all
            throw_on_error: Throw if download error occurres. Defaults to False.
            show_progress: Show progress bar in the console. Defaults to False.
            force_download: Download file even if it already exists locally. Defaults to False.

        Returns:
            A list of dictionaries with information about the downloaded files.
        """
        files = self.list_remote_files(project_id)

        return self.download_files(
            files,
            project_id,
            FileTransferType.PROJECT,
            local_dir,
            filter_glob,
            throw_on_error,
            show_progress,
            force_download,
        )

    def list_jobs(
        self,
        project_id: str,
        job_type: Optional[JobTypes] = None,
        pagination: Pagination = Pagination(),
    ) -> List[Dict[str, Any]]:
        """List project jobs.

        Args:
            project_id: Project ID.
            job_type: The type of job to filter by. If set to None, list all jobs. Defaults to None.
            pagination: Pagination settings. Defaults to a new Pagination object.

        Returns:
            A list of dictionaries representing the jobs.
        """
        payload = self._request_json(
            "GET",
            "jobs/",
            {
                "project_id": project_id,
                "type": job_type.value if job_type else None,
            },
            pagination=pagination,
        )
        return cast(List, payload)

    def job_trigger(
        self, project_id: str, job_type: JobTypes, force: bool = False
    ) -> Dict[str, Any]:
        """Trigger a new job for given project.

        Args:
            project_id: Project ID.
            job_type (JobTypes): The type of job to trigger.
            force: Whether to force the job execution. Defaults to False.

        Returns:
            A dictionary containing the job information.
        """
        resp = self._request(
            "POST",
            "jobs/",
            {
                "project_id": project_id,
                "type": job_type.value,
                "force": int(force),
            },
        )

        return resp.json()

    def job_status(self, job_id: str) -> Dict[str, Any]:
        """Get the status of a job.

        Args:
            job_id: The ID of the job.

        Returns:
            A dictionary containing the job status.
        """
        resp = self._request("GET", f"jobs/{job_id}")

        return resp.json()

    def delete_files(
        self,
        project_id: str,
        glob_patterns: List[str],
        throw_on_error: bool = False,
        finished_cb: Optional[Callable] = None,
    ) -> Dict[str, List[Dict[str, Any]]]:
        """Delete project files.

        Args:
            project_id: Project ID.
            glob_patterns (list[str]): Delete only files matching one the glob patterns.
            throw_on_error: Throw if delete error occurres. Defaults to False.
            finished_cb (Callable, optional): Deprecated. Defaults to None.

        Raises:
            QFieldCloudException: if throw_on_error is True, throw an error if a download request fails.

        Returns:
            Deleted files by glob pattern.
        """
        project_files = self.list_remote_files(project_id)
        glob_results: Dict[str, List[Dict[str, Any]]] = {}

        log(f"Project '{project_id}' has {len(project_files)} file(s).")

        for glob_pattern in glob_patterns:
            glob_results[glob_pattern] = []

            for file in project_files:
                if not fnmatch.fnmatch(file["name"], glob_pattern):
                    continue

                if "status" in file:
                    # file has already been matched by a previous glob pattern
                    continue

                file["status"] = FileTransferStatus.PENDING
                glob_results[glob_pattern].append(file)

        for glob_pattern, files in glob_results.items():
            if not files:
                log(f"Glob pattern '{glob_pattern}' did not match any files.")
                continue

            for file in files:
                try:
                    resp = self._request(
                        "DELETE",
                        f'files/{project_id}/{file["name"]}',
                        stream=True,
                    )
                    file["status"] = FileTransferStatus.SUCCESS
                except QfcRequestException as err:
                    resp = err.response

                    logger.info(
                        f"{resp.request.method} {resp.url} got HTTP {resp.status_code}"
                    )

                    file["status"] = FileTransferStatus.FAILED
                    file["error"] = err

                    log(f'File "{file["name"]}" failed to delete:\n{file["error"]}')

                    if throw_on_error:
                        continue
                    else:
                        raise err
                finally:
                    if callable(finished_cb):
                        finished_cb(file)

        files_deleted = 0
        files_failed = 0
        for files in glob_results.values():
            for file in files:
                log(f'{file["status"]}\t{file["name"]}')

                if file["status"] == FileTransferStatus.SUCCESS:
                    files_deleted += 1
                elif file["status"] == FileTransferStatus.SUCCESS:
                    files_failed += 1

        log(f"{files_deleted} file(s) deleted, {files_failed} file(s) failed to delete")

        return glob_results

    def package_latest(self, project_id: str) -> Dict[str, Any]:
        """Check the latest packaging status of a project.

        Args:
            project_id: Project ID.

        Returns:
            A dictionary containing the latest packaging status.
        """
        resp = self._request("GET", f"packages/{project_id}/latest/")

        return resp.json()

    def package_download(
        self,
        project_id: str,
        local_dir: str,
        filter_glob: str = "*",
        throw_on_error: bool = False,
        show_progress: bool = False,
        force_download: bool = False,
    ) -> List[Dict]:
        """Download the specified project packaged files into the destination dir.

        Args:
            project_id: Project ID.
            local_dir: destination directory where the files will be downloaded
            filter_glob: if specified, download only packaged files which match the glob, otherwise download all
            throw_on_error: Throw if download error occurres. Defaults to False.
            show_progress: Show progress bar in the console. Defaults to False.
            force_download: Download file even if it already exists locally. Defaults to False.

        Returns:
            A list of dictionaries with information about the downloaded files.
        """
        project_status = self.package_latest(project_id)

        if project_status["status"] != "finished":
            raise QfcException(
                "The project has not been successfully packaged yet. Please use `qfieldcloud-cli package-trigger {project_id}` first."
            )

        resp = self._request("GET", f"packages/{project_id}/latest/")

        return self.download_files(
            resp.json()["files"],
            project_id,
            FileTransferType.PACKAGE,
            local_dir,
            filter_glob,
            throw_on_error,
            show_progress,
            force_download,
        )

    def download_files(
        self,
        files: List[Dict],
        project_id: str,
        download_type: FileTransferType,
        local_dir: str,
        filter_glob: str = "*",
        throw_on_error: bool = False,
        show_progress: bool = False,
        force_download: bool = False,
    ) -> List[Dict]:
        """Download project files.

        Args:
            files : A list of file dicts, specifying which files to download.
            project_id: Project ID.
            download_type: File transfer type which specifies what should be the download url.
            local_dir: Local destination directory
            filter_glob: Download only files matching the glob pattern. If None download all. Defaults to None.
            throw_on_error: Throw if download error occurres. Defaults to False.
            show_progress: Show progress bar in the console. Defaults to False.
            force_download: Download file even if it already exists locally. Defaults to False.

        Raises:
            QFieldCloudException: if throw_on_error is True, throw an error if a download request fails.

        Returns:
            A list of file dicts.
        """
        if not filter_glob:
            filter_glob = "*"

        files_to_download: List[Dict[str, Any]] = []

        for file in files:
            if fnmatch.fnmatch(file["name"], filter_glob):
                file["status"] = FileTransferStatus.PENDING
                files_to_download.append(file)

        for file in files_to_download:
            local_filename = Path(f'{local_dir}/{file["name"]}')
            etag = None
            if not force_download:
                etag = file.get("etag", None)

            try:
                self.download_file(
                    project_id,
                    download_type,
                    local_filename,
                    file["name"],
                    show_progress,
                    etag,
                )
                file["status"] = FileTransferStatus.SUCCESS
            except QfcRequestException as err:
                resp = err.response

                logger.info(
                    f"{resp.request.method} {resp.url} got HTTP {resp.status_code}"
                )

                file["status"] = FileTransferStatus.FAILED
                file["error"] = err

                if throw_on_error:
                    raise err
                else:
                    continue

        return files_to_download

    def download_file(
        self,
        project_id: str,
        download_type: FileTransferType,
        local_filename: Path,
        remote_filename: Path,
        show_progress: bool,
        remote_etag: Optional[str] = None,
    ) -> Optional[requests.Response]:
        """Download a single project file.

        Args:
            project_id: Project ID.
            download_type: File transfer type which specifies what should be the download URL
            local_filename: Local filename
            remote_filename: Remote filename
            show_progress: Show progressbar in the console
            remote_etag: The ETag of the remote file. If is None, the download of the file happens even if it already exists locally. Defaults to `None`.

        Raises:
            NotImplementedError: Raised if unknown `download_type` is passed

        Returns:
            The response object.
        """

        if remote_etag and local_filename.exists():
            if calc_etag(str(local_filename)) == remote_etag:
                if show_progress:
                    print(
                        f"{remote_filename}: Already present locally. Download skipped."
                    )
                else:
                    logger.info(
                        f'Skipping download of "{remote_filename}" because it is already present locally'
                    )
                return None

        if download_type == FileTransferType.PROJECT:
            url = f"files/{project_id}/{remote_filename}"
        elif download_type == FileTransferType.PACKAGE:
            url = f"packages/{project_id}/latest/files/{remote_filename}"
        else:
            raise NotImplementedError()

        resp = self._request("GET", url, stream=True)

        if not local_filename.parent.exists():
            local_filename.parent.mkdir(parents=True)

        with open(local_filename, "wb") as f:
            download_file = f
            if show_progress:
                from tqdm import tqdm
                from tqdm.utils import CallbackIOWrapper

                content_length = int(resp.headers.get("content-length", 0))
                progress_bar = tqdm(
                    total=content_length,
                    unit_scale=True,
                    desc=remote_filename,
                )
                download_file = CallbackIOWrapper(progress_bar.update, f, "write")
            else:
                logger.info(f'Downloading file "{remote_filename}"…')

            for chunk in resp.iter_content(chunk_size=8192):
                # filter out keep-alive new chunks
                if chunk:
                    download_file.write(chunk)

        return resp

    def list_local_files(
        self, root_path: str, filter_glob: str
    ) -> List[Dict[str, Any]]:
        """
        Returns a list of dicts with information about local files. Usually used before uploading files.
        NOTE: files and dirs starting with leading dot (.) or ending in tilde (~) will be ignored.
        """
        if not filter_glob:
            filter_glob = "*"

        files: List[Dict[str, Any]] = []
        for path in Path(root_path).rglob(filter_glob):
            if not path.is_file():
                continue

            basename = path.relative_to(root_path).name
            if basename.startswith(".") or basename.endswith("~"):
                continue

            relative_name = path.relative_to(root_path)
            files.append(
                {
                    "name": str(relative_name),
                    "absolute_filename": str(path),
                    "status": FileTransferStatus.PENDING,
                    "error": None,
                }
            )

        # upload the QGIS project file at the end
        files.sort(key=lambda f: Path(f["name"]).suffix.lower() in (".qgs", ".qgz"))

        return files

    def get_project_collaborators(self, project_id: str) -> List[CollaboratorModel]:
        """Gets a list of project collaborators.

        Args:
            project_id: Project ID.

        Returns:
            The list of project collaborators.
        """
        collaborators = cast(
            List[CollaboratorModel],
            self._request_json("GET", f"/collaborators/{project_id}"),
        )

        return collaborators

    def add_project_collaborator(
        self, project_id: str, username: str, role: ProjectCollaboratorRole
    ) -> CollaboratorModel:
        """Adds a project collaborator.

        Args:
            project_id: Project ID.
            username: username of the collaborator to be added
            role: the role of the collaborator. One of: `reader`, `reporter`, `editor`, `manager` or `admin`

        Returns:
            The added project collaborator.
        """
        collaborator = cast(
            CollaboratorModel,
            self._request_json(
                "POST",
                f"/collaborators/{project_id}",
                {
                    "collaborator": username,
                    "role": role,
                },
            ),
        )

        return collaborator

    def remove_project_collaborator(self, project_id: str, username: str) -> None:
        """Removes a project collaborator.

        Args:
            project_id: Project ID.
            username: the username of the collaborator to be removed
        """
        self._request("DELETE", f"/collaborators/{project_id}/{username}")

    def patch_project_collaborators(
        self, project_id: str, username: str, role: ProjectCollaboratorRole
    ) -> CollaboratorModel:
        """Change an already existing project collaborator.

        Args:
            project_id: Project ID.
            username: the username of the collaborator to be patched
            role: the new role of the collaborator

        Returns:
            The updated project collaborator.
        """
        collaborator = cast(
            CollaboratorModel,
            self._request_json(
                "PATCH",
                f"/collaborators/{project_id}/{username}",
                {
                    "role": role,
                },
            ),
        )

        return collaborator

    def get_organization_members(
        self, organization: str
    ) -> List[OrganizationMemberModel]:
        """Gets a list of organization members.

        Args:
            organization: organization username

        Returns:
            The list of organization members.
        """
        members = cast(
            List[OrganizationMemberModel],
            self._request_json("GET", f"/members/{organization}"),
        )

        return members

    def add_organization_member(
        self,
        organization: str,
        username: str,
        role: OrganizationMemberRole,
        is_public: bool,
    ) -> OrganizationMemberModel:
        """Adds an organization member.

        Args:
            organization: organization username
            username: username of the member to be added
            role: the role of the member. One of: `admin` or `member`.
            is_public: whether the membership should be public.

        Returns:
            The added organization member.
        """
        member = cast(
            OrganizationMemberModel,
            self._request_json(
                "POST",
                f"/members/{organization}",
                {
                    "member": username,
                    "role": role,
                    "is_public": is_public,
                },
            ),
        )

        return member

    def remove_organization_members(self, project_id: str, username: str) -> None:
        """Removes an organization member.

        Args:
            project_id: Project ID.
            username: the username of the member to be removed
        """
        self._request("DELETE", f"/members/{project_id}/{username}")

    def patch_organization_members(
        self, project_id: str, username: str, role: OrganizationMemberRole
    ) -> OrganizationMemberModel:
        """Change an already existing organization member.

        Args:
            project_id: Project ID.
            username: the username of the member to be patched
            role: the new role of the member

        Returns:
            The updated organization member.
        """
        member = cast(
            OrganizationMemberModel,
            self._request_json(
                "PATCH",
                f"/members/{project_id}/{username}",
                {
                    "role": role,
                },
            ),
        )

        return member

    def _request_json(
        self,
        method: str,
        path: str,
        data: Any = None,
        params: Dict[str, str] = {},
        headers: Dict[str, str] = {},
        files: Optional[Dict[str, Any]] = None,
        stream: bool = False,
        skip_token: bool = False,
        allow_redirects=None,
        pagination: Pagination = Pagination(),
    ) -> Union[List, Dict]:
        result: Optional[Union[List, Dict]] = None
        is_empty_pagination = pagination.is_empty

        while True:
            resp = self._request(
                method,
                path,
                data,
                params,
                headers,
                files,
                stream,
                skip_token,
                allow_redirects,
                pagination,
            )

            payload = resp.json()

            if isinstance(payload, list):
                result = cast(List, result)

                if result:
                    result += payload
                else:
                    result = payload
            elif isinstance(payload, dict):
                result = cast(Dict, result)

                if result:
                    result = {**result, **payload}
                else:
                    result = payload
            else:
                raise NotImplementedError(
                    "Unsupported data type for paginated response."
                )

            if not is_empty_pagination:
                break

            next_url = resp.headers.get("X-Next-Page")
            if not next_url:
                break

            query_params = urlparse.parse_qs(urlparse.urlparse(next_url).query)
            pagination = Pagination(
                limit=cast(int, query_params["limit"]),
                offset=cast(int, query_params["offset"]),
            )

        return result

    def _request(
        self,
        method: str,
        path: str,
        data: Any = None,
        params: Dict[str, str] = {},
        headers: Dict[str, str] = {},
        files: Optional[Dict[str, Any]] = None,
        stream: bool = False,
        skip_token: bool = False,
        allow_redirects=None,
        pagination: Optional[Pagination] = None,
    ) -> requests.Response:
        method = method.upper()
        headers_copy = {**headers}

        assert self.url

        allow_redirects = method != "POST"

        if not skip_token and self.token:
            headers_copy["Authorization"] = f"token {self.token}"

        headers_copy["User-Agent"] = (
            f"sdk|py|{__version__} python-requests|{metadata.version('requests')}"
        )

        if not path.startswith("http"):
            if path.startswith("/"):
                path = path[1:]

            if not path.endswith("/"):
                path += "/"

            path = self.url + path

        if pagination:
            limit = pagination.limit or DEFAULT_PAGINATION_LIMIT
            offset = pagination.offset or 0
            params = {
                **params,
                "limit": str(limit),
                "offset": str(offset),
            }

        request_params = {
            "method": method,
            "url": path,
            "data": data,
            "params": params,
            "headers": headers_copy,
            "files": files,
        }

        request = QfcRequest(**request_params)

        session_params = {
            "stream": stream,
            "verify": self.verify_ssl,
            # redirects from POST requests automagically turn into GET requests, so better forbid redirects
            "allow_redirects": allow_redirects,
        }

        if os.environ.get("ENVIRONMENT") == "test":
            return request.mock_response()
        else:
            response = self.session.send(request.prepare(), **session_params)

        try:
            response.raise_for_status()
        except Exception as err:
            raise QfcRequestException(response) from err

        return response

add_organization_member(organization, username, role, is_public)

Adds an organization member.

Parameters:

Name	Type	Description	Default
organization	str	organization username	required
username	str	username of the member to be added	required
role	OrganizationMemberRole	the role of the member. One of: admin or member.	required
is_public	bool	whether the membership should be public.	required

Returns:

Type	Description
OrganizationMemberModel	The added organization member.

Source code in qfieldcloud_sdk/sdk.py

def add_organization_member(
    self,
    organization: str,
    username: str,
    role: OrganizationMemberRole,
    is_public: bool,
) -> OrganizationMemberModel:
    """Adds an organization member.

    Args:
        organization: organization username
        username: username of the member to be added
        role: the role of the member. One of: `admin` or `member`.
        is_public: whether the membership should be public.

    Returns:
        The added organization member.
    """
    member = cast(
        OrganizationMemberModel,
        self._request_json(
            "POST",
            f"/members/{organization}",
            {
                "member": username,
                "role": role,
                "is_public": is_public,
            },
        ),
    )

    return member

add_project_collaborator(project_id, username, role)

Adds a project collaborator.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
username	str	username of the collaborator to be added	required
role	ProjectCollaboratorRole	the role of the collaborator. One of: reader, reporter, editor, manager or admin	required

Returns:

Type	Description
CollaboratorModel	The added project collaborator.

Source code in qfieldcloud_sdk/sdk.py

def add_project_collaborator(
    self, project_id: str, username: str, role: ProjectCollaboratorRole
) -> CollaboratorModel:
    """Adds a project collaborator.

    Args:
        project_id: Project ID.
        username: username of the collaborator to be added
        role: the role of the collaborator. One of: `reader`, `reporter`, `editor`, `manager` or `admin`

    Returns:
        The added project collaborator.
    """
    collaborator = cast(
        CollaboratorModel,
        self._request_json(
            "POST",
            f"/collaborators/{project_id}",
            {
                "collaborator": username,
                "role": role,
            },
        ),
    )

    return collaborator

create_project(name, owner=None, description='', is_public=False)

Create a new project.

Parameters:

Name	Type	Description	Default
name	str	The name of the new project.	required
owner	Optional[str]	The owner of the project. When None, the project will be owned by the currently logged-in user. Defaults to None.	None
description	str	A description of the project. Defaults to an empty string.	''
is_public	bool	Whether the project should be public. Defaults to False.	False

Returns:

Type	Description
Dict[str, Any]	A dictionary containing the details of the created project.

Source code in qfieldcloud_sdk/sdk.py

def create_project(
    self,
    name: str,
    owner: Optional[str] = None,
    description: str = "",
    is_public: bool = False,
) -> Dict[str, Any]:
    """Create a new project.

    Args:
        name: The name of the new project.
        owner: The owner of the project. When None, the project will be owned by the currently logged-in user. Defaults to None.
        description: A description of the project. Defaults to an empty string.
        is_public: Whether the project should be public. Defaults to False.

    Returns:
        A dictionary containing the details of the created project.
    """
    resp = self._request(
        "POST",
        "projects",
        data={
            "name": name,
            "owner": owner,
            "description": description,
            "is_public": int(is_public),
        },
    )

    return resp.json()

delete_files(project_id, glob_patterns, throw_on_error=False, finished_cb=None)

Delete project files.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
glob_patterns	list[str]	Delete only files matching one the glob patterns.	required
throw_on_error	bool	Throw if delete error occurres. Defaults to False.	False
finished_cb	Callable	Deprecated. Defaults to None.	None

Raises:

Type	Description
QFieldCloudException	if throw_on_error is True, throw an error if a download request fails.

Returns:

Type	Description
Dict[str, List[Dict[str, Any]]]	Deleted files by glob pattern.

Source code in qfieldcloud_sdk/sdk.py

def delete_files(
    self,
    project_id: str,
    glob_patterns: List[str],
    throw_on_error: bool = False,
    finished_cb: Optional[Callable] = None,
) -> Dict[str, List[Dict[str, Any]]]:
    """Delete project files.

    Args:
        project_id: Project ID.
        glob_patterns (list[str]): Delete only files matching one the glob patterns.
        throw_on_error: Throw if delete error occurres. Defaults to False.
        finished_cb (Callable, optional): Deprecated. Defaults to None.

    Raises:
        QFieldCloudException: if throw_on_error is True, throw an error if a download request fails.

    Returns:
        Deleted files by glob pattern.
    """
    project_files = self.list_remote_files(project_id)
    glob_results: Dict[str, List[Dict[str, Any]]] = {}

    log(f"Project '{project_id}' has {len(project_files)} file(s).")

    for glob_pattern in glob_patterns:
        glob_results[glob_pattern] = []

        for file in project_files:
            if not fnmatch.fnmatch(file["name"], glob_pattern):
                continue

            if "status" in file:
                # file has already been matched by a previous glob pattern
                continue

            file["status"] = FileTransferStatus.PENDING
            glob_results[glob_pattern].append(file)

    for glob_pattern, files in glob_results.items():
        if not files:
            log(f"Glob pattern '{glob_pattern}' did not match any files.")
            continue

        for file in files:
            try:
                resp = self._request(
                    "DELETE",
                    f'files/{project_id}/{file["name"]}',
                    stream=True,
                )
                file["status"] = FileTransferStatus.SUCCESS
            except QfcRequestException as err:
                resp = err.response

                logger.info(
                    f"{resp.request.method} {resp.url} got HTTP {resp.status_code}"
                )

                file["status"] = FileTransferStatus.FAILED
                file["error"] = err

                log(f'File "{file["name"]}" failed to delete:\n{file["error"]}')

                if throw_on_error:
                    continue
                else:
                    raise err
            finally:
                if callable(finished_cb):
                    finished_cb(file)

    files_deleted = 0
    files_failed = 0
    for files in glob_results.values():
        for file in files:
            log(f'{file["status"]}\t{file["name"]}')

            if file["status"] == FileTransferStatus.SUCCESS:
                files_deleted += 1
            elif file["status"] == FileTransferStatus.SUCCESS:
                files_failed += 1

    log(f"{files_deleted} file(s) deleted, {files_failed} file(s) failed to delete")

    return glob_results

delete_project(project_id)

Delete a project.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required

Returns:

Type	Description
Response	The response object from the file delete request.

Source code in qfieldcloud_sdk/sdk.py

def delete_project(self, project_id: str) -> requests.Response:
    """Delete a project.

    Args:
        project_id: Project ID.

    Returns:
        The response object from the file delete request.
    """
    resp = self._request("DELETE", f"projects/{project_id}")

    return resp

download_file(project_id, download_type, local_filename, remote_filename, show_progress, remote_etag=None)

Download a single project file.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
download_type	FileTransferType	File transfer type which specifies what should be the download URL	required
local_filename	Path	Local filename	required
remote_filename	Path	Remote filename	required
show_progress	bool	Show progressbar in the console	required
remote_etag	Optional[str]	The ETag of the remote file. If is None, the download of the file happens even if it already exists locally. Defaults to None.	None

Raises:

Type	Description
NotImplementedError	Raised if unknown download_type is passed

Returns:

Type	Description
Optional[Response]	The response object.

Source code in qfieldcloud_sdk/sdk.py

def download_file(
    self,
    project_id: str,
    download_type: FileTransferType,
    local_filename: Path,
    remote_filename: Path,
    show_progress: bool,
    remote_etag: Optional[str] = None,
) -> Optional[requests.Response]:
    """Download a single project file.

    Args:
        project_id: Project ID.
        download_type: File transfer type which specifies what should be the download URL
        local_filename: Local filename
        remote_filename: Remote filename
        show_progress: Show progressbar in the console
        remote_etag: The ETag of the remote file. If is None, the download of the file happens even if it already exists locally. Defaults to `None`.

    Raises:
        NotImplementedError: Raised if unknown `download_type` is passed

    Returns:
        The response object.
    """

    if remote_etag and local_filename.exists():
        if calc_etag(str(local_filename)) == remote_etag:
            if show_progress:
                print(
                    f"{remote_filename}: Already present locally. Download skipped."
                )
            else:
                logger.info(
                    f'Skipping download of "{remote_filename}" because it is already present locally'
                )
            return None

    if download_type == FileTransferType.PROJECT:
        url = f"files/{project_id}/{remote_filename}"
    elif download_type == FileTransferType.PACKAGE:
        url = f"packages/{project_id}/latest/files/{remote_filename}"
    else:
        raise NotImplementedError()

    resp = self._request("GET", url, stream=True)

    if not local_filename.parent.exists():
        local_filename.parent.mkdir(parents=True)

    with open(local_filename, "wb") as f:
        download_file = f
        if show_progress:
            from tqdm import tqdm
            from tqdm.utils import CallbackIOWrapper

            content_length = int(resp.headers.get("content-length", 0))
            progress_bar = tqdm(
                total=content_length,
                unit_scale=True,
                desc=remote_filename,
            )
            download_file = CallbackIOWrapper(progress_bar.update, f, "write")
        else:
            logger.info(f'Downloading file "{remote_filename}"…')

        for chunk in resp.iter_content(chunk_size=8192):
            # filter out keep-alive new chunks
            if chunk:
                download_file.write(chunk)

    return resp

download_files(files, project_id, download_type, local_dir, filter_glob='*', throw_on_error=False, show_progress=False, force_download=False)

Download project files.

Parameters:

Name	Type	Description	Default
files		A list of file dicts, specifying which files to download.	required
project_id	str	Project ID.	required
download_type	FileTransferType	File transfer type which specifies what should be the download url.	required
local_dir	str	Local destination directory	required
filter_glob	str	Download only files matching the glob pattern. If None download all. Defaults to None.	'*'
throw_on_error	bool	Throw if download error occurres. Defaults to False.	False
show_progress	bool	Show progress bar in the console. Defaults to False.	False
force_download	bool	Download file even if it already exists locally. Defaults to False.	False

Raises:

Type	Description
QFieldCloudException	if throw_on_error is True, throw an error if a download request fails.

Returns:

Type	Description
List[Dict]	A list of file dicts.

Source code in qfieldcloud_sdk/sdk.py

def download_files(
    self,
    files: List[Dict],
    project_id: str,
    download_type: FileTransferType,
    local_dir: str,
    filter_glob: str = "*",
    throw_on_error: bool = False,
    show_progress: bool = False,
    force_download: bool = False,
) -> List[Dict]:
    """Download project files.

    Args:
        files : A list of file dicts, specifying which files to download.
        project_id: Project ID.
        download_type: File transfer type which specifies what should be the download url.
        local_dir: Local destination directory
        filter_glob: Download only files matching the glob pattern. If None download all. Defaults to None.
        throw_on_error: Throw if download error occurres. Defaults to False.
        show_progress: Show progress bar in the console. Defaults to False.
        force_download: Download file even if it already exists locally. Defaults to False.

    Raises:
        QFieldCloudException: if throw_on_error is True, throw an error if a download request fails.

    Returns:
        A list of file dicts.
    """
    if not filter_glob:
        filter_glob = "*"

    files_to_download: List[Dict[str, Any]] = []

    for file in files:
        if fnmatch.fnmatch(file["name"], filter_glob):
            file["status"] = FileTransferStatus.PENDING
            files_to_download.append(file)

    for file in files_to_download:
        local_filename = Path(f'{local_dir}/{file["name"]}')
        etag = None
        if not force_download:
            etag = file.get("etag", None)

        try:
            self.download_file(
                project_id,
                download_type,
                local_filename,
                file["name"],
                show_progress,
                etag,
            )
            file["status"] = FileTransferStatus.SUCCESS
        except QfcRequestException as err:
            resp = err.response

            logger.info(
                f"{resp.request.method} {resp.url} got HTTP {resp.status_code}"
            )

            file["status"] = FileTransferStatus.FAILED
            file["error"] = err

            if throw_on_error:
                raise err
            else:
                continue

    return files_to_download

download_project(project_id, local_dir, filter_glob='*', throw_on_error=False, show_progress=False, force_download=False)

Download the specified project files into the destination dir.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
local_dir	str	destination directory where the files will be downloaded	required
filter_glob	str	if specified, download only project files which match the glob, otherwise download all	'*'
throw_on_error	bool	Throw if download error occurres. Defaults to False.	False
show_progress	bool	Show progress bar in the console. Defaults to False.	False
force_download	bool	Download file even if it already exists locally. Defaults to False.	False

Returns:

Type	Description
List[Dict]	A list of dictionaries with information about the downloaded files.

Source code in qfieldcloud_sdk/sdk.py

def download_project(
    self,
    project_id: str,
    local_dir: str,
    filter_glob: str = "*",
    throw_on_error: bool = False,
    show_progress: bool = False,
    force_download: bool = False,
) -> List[Dict]:
    """Download the specified project files into the destination dir.

    Args:
        project_id: Project ID.
        local_dir: destination directory where the files will be downloaded
        filter_glob: if specified, download only project files which match the glob, otherwise download all
        throw_on_error: Throw if download error occurres. Defaults to False.
        show_progress: Show progress bar in the console. Defaults to False.
        force_download: Download file even if it already exists locally. Defaults to False.

    Returns:
        A list of dictionaries with information about the downloaded files.
    """
    files = self.list_remote_files(project_id)

    return self.download_files(
        files,
        project_id,
        FileTransferType.PROJECT,
        local_dir,
        filter_glob,
        throw_on_error,
        show_progress,
        force_download,
    )

get_organization_members(organization)

Gets a list of organization members.

Parameters:

Name	Type	Description	Default
organization	str	organization username	required

Returns:

Type	Description
List[OrganizationMemberModel]	The list of organization members.

Source code in qfieldcloud_sdk/sdk.py

def get_organization_members(
    self, organization: str
) -> List[OrganizationMemberModel]:
    """Gets a list of organization members.

    Args:
        organization: organization username

    Returns:
        The list of organization members.
    """
    members = cast(
        List[OrganizationMemberModel],
        self._request_json("GET", f"/members/{organization}"),
    )

    return members

get_project_collaborators(project_id)

Gets a list of project collaborators.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required

Returns:

Type	Description
List[CollaboratorModel]	The list of project collaborators.

Source code in qfieldcloud_sdk/sdk.py

def get_project_collaborators(self, project_id: str) -> List[CollaboratorModel]:
    """Gets a list of project collaborators.

    Args:
        project_id: Project ID.

    Returns:
        The list of project collaborators.
    """
    collaborators = cast(
        List[CollaboratorModel],
        self._request_json("GET", f"/collaborators/{project_id}"),
    )

    return collaborators

job_status(job_id)

Get the status of a job.

Parameters:

Name	Type	Description	Default
job_id	str	The ID of the job.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary containing the job status.

Source code in qfieldcloud_sdk/sdk.py

def job_status(self, job_id: str) -> Dict[str, Any]:
    """Get the status of a job.

    Args:
        job_id: The ID of the job.

    Returns:
        A dictionary containing the job status.
    """
    resp = self._request("GET", f"jobs/{job_id}")

    return resp.json()

job_trigger(project_id, job_type, force=False)

Trigger a new job for given project.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
job_type	JobTypes	The type of job to trigger.	required
force	bool	Whether to force the job execution. Defaults to False.	False

Returns:

Type	Description
Dict[str, Any]	A dictionary containing the job information.

Source code in qfieldcloud_sdk/sdk.py

def job_trigger(
    self, project_id: str, job_type: JobTypes, force: bool = False
) -> Dict[str, Any]:
    """Trigger a new job for given project.

    Args:
        project_id: Project ID.
        job_type (JobTypes): The type of job to trigger.
        force: Whether to force the job execution. Defaults to False.

    Returns:
        A dictionary containing the job information.
    """
    resp = self._request(
        "POST",
        "jobs/",
        {
            "project_id": project_id,
            "type": job_type.value,
            "force": int(force),
        },
    )

    return resp.json()

list_jobs(project_id, job_type=None, pagination=Pagination())

List project jobs.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
job_type	Optional[JobTypes]	The type of job to filter by. If set to None, list all jobs. Defaults to None.	None
pagination	Pagination	Pagination settings. Defaults to a new Pagination object.	Pagination()

Returns:

Type	Description
List[Dict[str, Any]]	A list of dictionaries representing the jobs.

Source code in qfieldcloud_sdk/sdk.py

def list_jobs(
    self,
    project_id: str,
    job_type: Optional[JobTypes] = None,
    pagination: Pagination = Pagination(),
) -> List[Dict[str, Any]]:
    """List project jobs.

    Args:
        project_id: Project ID.
        job_type: The type of job to filter by. If set to None, list all jobs. Defaults to None.
        pagination: Pagination settings. Defaults to a new Pagination object.

    Returns:
        A list of dictionaries representing the jobs.
    """
    payload = self._request_json(
        "GET",
        "jobs/",
        {
            "project_id": project_id,
            "type": job_type.value if job_type else None,
        },
        pagination=pagination,
    )
    return cast(List, payload)

list_local_files(root_path, filter_glob)

Returns a list of dicts with information about local files. Usually used before uploading files. NOTE: files and dirs starting with leading dot (.) or ending in tilde (~) will be ignored.

Source code in qfieldcloud_sdk/sdk.py

def list_local_files(
    self, root_path: str, filter_glob: str
) -> List[Dict[str, Any]]:
    """
    Returns a list of dicts with information about local files. Usually used before uploading files.
    NOTE: files and dirs starting with leading dot (.) or ending in tilde (~) will be ignored.
    """
    if not filter_glob:
        filter_glob = "*"

    files: List[Dict[str, Any]] = []
    for path in Path(root_path).rglob(filter_glob):
        if not path.is_file():
            continue

        basename = path.relative_to(root_path).name
        if basename.startswith(".") or basename.endswith("~"):
            continue

        relative_name = path.relative_to(root_path)
        files.append(
            {
                "name": str(relative_name),
                "absolute_filename": str(path),
                "status": FileTransferStatus.PENDING,
                "error": None,
            }
        )

    # upload the QGIS project file at the end
    files.sort(key=lambda f: Path(f["name"]).suffix.lower() in (".qgs", ".qgz"))

    return files

list_projects(include_public=False, pagination=Pagination(), **kwargs)

List projects accessible by the current user. Optionally include all public projects.

Parameters:

Name	Type	Description	Default
include_public	bool	Whether to include public projects in the list. Defaults to False.	False
pagination	Pagination	Pagination settings for the request. Defaults to an empty Pagination instance.	Pagination()

Returns:

Type	Description
List[Dict[str, Any]]	A list of dictionaries containing project details.

Source code in qfieldcloud_sdk/sdk.py

def list_projects(
    self,
    include_public: bool = False,
    pagination: Pagination = Pagination(),
    **kwargs,
) -> List[Dict[str, Any]]:
    """List projects accessible by the current user. Optionally include all public projects.

    Args:
        include_public: Whether to include public projects in the list. Defaults to False.
        pagination: Pagination settings for the request. Defaults to an empty Pagination instance.

    Returns:
        A list of dictionaries containing project details.
    """
    params = {
        "include-public": str(int(include_public)),  # type: ignore
    }

    payload = self._request_json(
        "GET", "projects", params=params, pagination=pagination
    )
    return cast(List, payload)

list_remote_files(project_id, skip_metadata=True)

List project files.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
skip_metadata	bool	Whether to skip fetching metadata for the files. Defaults to True.	True

Returns:

Type	Description
List[Dict[str, Any]]	A list of file details.

Example

client.list_remote_files("project_id", True)

Source code in qfieldcloud_sdk/sdk.py

def list_remote_files(
    self, project_id: str, skip_metadata: bool = True
) -> List[Dict[str, Any]]:
    """List project files.

    Args:
        project_id: Project ID.
        skip_metadata: Whether to skip fetching metadata for the files. Defaults to True.

    Returns:
        A list of file details.

    Example:
        client.list_remote_files("project_id", True)
    """
    params = {}

    if skip_metadata:
        params["skip_metadata"] = "1"

    resp = self._request("GET", f"files/{project_id}", params=params)
    remote_files = resp.json()
    # TODO remove this temporary decoration with `etag` key
    remote_files = list(map(lambda f: {"etag": f["md5sum"], **f}, remote_files))

    return remote_files

login(username, password)

Logs in with the provided username and password.

Parameters:

Name	Type	Description	Default
username	str	The username or email used to register.	required
password	str	The password associated with the username.	required

Returns:

Type	Description
Dict[str, Any]	Authentication token and additional metadata.

Example

client = sdk.Client(url="https://app.qfield.cloud/api/v1/") client.login("ninjamaster", "secret_password123")

Source code in qfieldcloud_sdk/sdk.py

def login(self, username: str, password: str) -> Dict[str, Any]:
    """Logs in with the provided username and password.

    Args:
        username: The username or email used to register.
        password: The password associated with the username.

    Returns:
        Authentication token and additional metadata.

    Example:
        client = sdk.Client(url="https://app.qfield.cloud/api/v1/")
        client.login("ninjamaster", "secret_password123")
    """
    resp = self._request(
        "POST",
        "auth/login",
        data={
            "username": username,
            "password": password,
        },
        skip_token=True,
    )

    payload = resp.json()

    self.token = payload["token"]

    return payload

logout()

Logs out from the current session, invalidating the authentication token.

Example

client.logout()

Source code in qfieldcloud_sdk/sdk.py

def logout(self) -> None:
    """Logs out from the current session, invalidating the authentication token.

    Example:
        client.logout()
    """
    resp = self._request("POST", "auth/logout")

    return resp.json()

package_download(project_id, local_dir, filter_glob='*', throw_on_error=False, show_progress=False, force_download=False)

Download the specified project packaged files into the destination dir.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
local_dir	str	destination directory where the files will be downloaded	required
filter_glob	str	if specified, download only packaged files which match the glob, otherwise download all	'*'
throw_on_error	bool	Throw if download error occurres. Defaults to False.	False
show_progress	bool	Show progress bar in the console. Defaults to False.	False
force_download	bool	Download file even if it already exists locally. Defaults to False.	False

Returns:

Type	Description
List[Dict]	A list of dictionaries with information about the downloaded files.

Source code in qfieldcloud_sdk/sdk.py

def package_download(
    self,
    project_id: str,
    local_dir: str,
    filter_glob: str = "*",
    throw_on_error: bool = False,
    show_progress: bool = False,
    force_download: bool = False,
) -> List[Dict]:
    """Download the specified project packaged files into the destination dir.

    Args:
        project_id: Project ID.
        local_dir: destination directory where the files will be downloaded
        filter_glob: if specified, download only packaged files which match the glob, otherwise download all
        throw_on_error: Throw if download error occurres. Defaults to False.
        show_progress: Show progress bar in the console. Defaults to False.
        force_download: Download file even if it already exists locally. Defaults to False.

    Returns:
        A list of dictionaries with information about the downloaded files.
    """
    project_status = self.package_latest(project_id)

    if project_status["status"] != "finished":
        raise QfcException(
            "The project has not been successfully packaged yet. Please use `qfieldcloud-cli package-trigger {project_id}` first."
        )

    resp = self._request("GET", f"packages/{project_id}/latest/")

    return self.download_files(
        resp.json()["files"],
        project_id,
        FileTransferType.PACKAGE,
        local_dir,
        filter_glob,
        throw_on_error,
        show_progress,
        force_download,
    )

package_latest(project_id)

Check the latest packaging status of a project.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary containing the latest packaging status.

Source code in qfieldcloud_sdk/sdk.py

def package_latest(self, project_id: str) -> Dict[str, Any]:
    """Check the latest packaging status of a project.

    Args:
        project_id: Project ID.

    Returns:
        A dictionary containing the latest packaging status.
    """
    resp = self._request("GET", f"packages/{project_id}/latest/")

    return resp.json()

patch_organization_members(project_id, username, role)

Change an already existing organization member.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
username	str	the username of the member to be patched	required
role	OrganizationMemberRole	the new role of the member	required

Returns:

Type	Description
OrganizationMemberModel	The updated organization member.

Source code in qfieldcloud_sdk/sdk.py

def patch_organization_members(
    self, project_id: str, username: str, role: OrganizationMemberRole
) -> OrganizationMemberModel:
    """Change an already existing organization member.

    Args:
        project_id: Project ID.
        username: the username of the member to be patched
        role: the new role of the member

    Returns:
        The updated organization member.
    """
    member = cast(
        OrganizationMemberModel,
        self._request_json(
            "PATCH",
            f"/members/{project_id}/{username}",
            {
                "role": role,
            },
        ),
    )

    return member

patch_project_collaborators(project_id, username, role)

Change an already existing project collaborator.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
username	str	the username of the collaborator to be patched	required
role	ProjectCollaboratorRole	the new role of the collaborator	required

Returns:

Type	Description
CollaboratorModel	The updated project collaborator.

Source code in qfieldcloud_sdk/sdk.py

def patch_project_collaborators(
    self, project_id: str, username: str, role: ProjectCollaboratorRole
) -> CollaboratorModel:
    """Change an already existing project collaborator.

    Args:
        project_id: Project ID.
        username: the username of the collaborator to be patched
        role: the new role of the collaborator

    Returns:
        The updated project collaborator.
    """
    collaborator = cast(
        CollaboratorModel,
        self._request_json(
            "PATCH",
            f"/collaborators/{project_id}/{username}",
            {
                "role": role,
            },
        ),
    )

    return collaborator

remove_organization_members(project_id, username)

Removes an organization member.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
username	str	the username of the member to be removed	required

Source code in qfieldcloud_sdk/sdk.py

def remove_organization_members(self, project_id: str, username: str) -> None:
    """Removes an organization member.

    Args:
        project_id: Project ID.
        username: the username of the member to be removed
    """
    self._request("DELETE", f"/members/{project_id}/{username}")

remove_project_collaborator(project_id, username)

Removes a project collaborator.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
username	str	the username of the collaborator to be removed	required

Source code in qfieldcloud_sdk/sdk.py

def remove_project_collaborator(self, project_id: str, username: str) -> None:
    """Removes a project collaborator.

    Args:
        project_id: Project ID.
        username: the username of the collaborator to be removed
    """
    self._request("DELETE", f"/collaborators/{project_id}/{username}")

upload_file(project_id, upload_type, local_filename, remote_filename, show_progress, job_id='')

Upload a single file to a project.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
upload_type	FileTransferType	The type of file transfer.	required
local_filename	Path	The path to the local file to upload.	required
remote_filename	Path	The path where the file should be stored remotely.	required
show_progress	bool	Whether to display a progress bar during upload.	required
job_id	str	The job ID, required if upload_type is PACKAGE. Defaults to an empty string.	''

Returns:

Type	Description
Response	The response object from the upload request.

Source code in qfieldcloud_sdk/sdk.py

def upload_file(
    self,
    project_id: str,
    upload_type: FileTransferType,
    local_filename: Path,
    remote_filename: Path,
    show_progress: bool,
    job_id: str = "",
) -> requests.Response:
    """Upload a single file to a project.

    Args:
        project_id: Project ID.
        upload_type: The type of file transfer.
        local_filename: The path to the local file to upload.
        remote_filename: The path where the file should be stored remotely.
        show_progress: Whether to display a progress bar during upload.
        job_id: The job ID, required if `upload_type` is PACKAGE. Defaults to an empty string.

    Returns:
        The response object from the upload request.
    """
    with open(local_filename, "rb") as local_file:
        upload_file = local_file
        if show_progress:
            from tqdm import tqdm
            from tqdm.utils import CallbackIOWrapper

            progress_bar = tqdm(
                total=local_filename.stat().st_size,
                unit_scale=True,
                desc=local_filename.stem,
            )
            upload_file = CallbackIOWrapper(progress_bar.update, local_file, "read")
        else:
            logger.info(f'Uploading file "{remote_filename}"…')

        if upload_type == FileTransferType.PROJECT:
            url = f"files/{project_id}/{remote_filename}"
        elif upload_type == FileTransferType.PACKAGE:
            if not job_id:
                raise QfcException(
                    'When the upload type is "package", you must pass the "job_id" parameter.'
                )

            url = f"packages/{project_id}/{job_id}/files/{remote_filename}"
        else:
            raise NotImplementedError()

        return self._request(
            "POST",
            url,
            files={
                "file": upload_file,
            },
        )

upload_files(project_id, upload_type, project_path, filter_glob, throw_on_error=False, show_progress=False, force=False, job_id='')

Upload files to a QFieldCloud project.

Parameters:

Name	Type	Description	Default
project_id	str	Project ID.	required
upload_type	FileTransferType	The type of file transfer (PROJECT or PACKAGE).	required
project_path	str	The local directory containing the files to upload.	required
filter_glob	str	A glob pattern to filter which files to upload.	required
throw_on_error	bool	Whether to raise an error if a file fails to upload. Defaults to False.	False
show_progress	bool	Whether to display a progress bar during upload. Defaults to False.	False
force	bool	Whether to force upload all files, even if they exist remotely. Defaults to False.	False
job_id	str	The job ID, required if upload_type is PACKAGE. Defaults to an empty string.	''

Returns:

Type	Description
List[Dict]	A list of dictionaries with information about the uploaded files.

Source code in qfieldcloud_sdk/sdk.py

def upload_files(
    self,
    project_id: str,
    upload_type: FileTransferType,
    project_path: str,
    filter_glob: str,
    throw_on_error: bool = False,
    show_progress: bool = False,
    force: bool = False,
    job_id: str = "",
) -> List[Dict]:
    """Upload files to a QFieldCloud project.

    Args:
        project_id: Project ID.
        upload_type: The type of file transfer (PROJECT or PACKAGE).
        project_path: The local directory containing the files to upload.
        filter_glob: A glob pattern to filter which files to upload.
        throw_on_error: Whether to raise an error if a file fails to upload. Defaults to False.
        show_progress: Whether to display a progress bar during upload. Defaults to False.
        force: Whether to force upload all files, even if they exist remotely. Defaults to False.
        job_id: The job ID, required if `upload_type` is PACKAGE. Defaults to an empty string.

    Returns:
        A list of dictionaries with information about the uploaded files.
    """
    if not filter_glob:
        filter_glob = "*"

    local_files = self.list_local_files(project_path, filter_glob)

    # we should always upload all package files
    if upload_type == FileTransferType.PACKAGE:
        force = True

    files_to_upload = []
    if force:
        files_to_upload = local_files
    else:
        remote_files = self.list_remote_files(project_id)

        if len(remote_files) == 0:
            files_to_upload = local_files
        else:
            for local_file in local_files:
                remote_file = None
                for f in remote_files:
                    if f["name"] == local_file["name"]:
                        remote_file = f
                        break

                etag = calc_etag(local_file["absolute_filename"])
                if remote_file and remote_file.get("etag", None) == etag:
                    continue

                files_to_upload.append(local_file)

    if not files_to_upload:
        return files_to_upload

    for file in files_to_upload:
        try:
            local_filename = Path(file["absolute_filename"])
            self.upload_file(
                project_id,
                upload_type,
                local_filename,
                file["name"],
                show_progress,
                job_id,
            )
            file["status"] = FileTransferStatus.SUCCESS
        except Exception as err:
            file["status"] = FileTransferStatus.FAILED
            file["error"] = err

            if throw_on_error:
                raise err
            else:
                continue

    return local_files

CollaboratorModel

Bases: TypedDict

Represents the structure of a project collaborator in the QFieldCloud system.

Attributes:

Name	Type	Description
collaborator	str	The collaborator's identifier.
role	ProjectCollaboratorRole	The role of the collaborator.
project_id	str	The associated project identifier.
created_by	str	The user who created the collaborator entry.
updated_by	str	The user who last updated the collaborator entry.
created_at	datetime	The timestamp when the collaborator entry was created.
updated_at	datetime	The timestamp when the collaborator entry was last updated.

Source code in qfieldcloud_sdk/sdk.py

class CollaboratorModel(TypedDict):
    """Represents the structure of a project collaborator in the QFieldCloud system.

    Attributes:
        collaborator: The collaborator's identifier.
        role: The role of the collaborator.
        project_id: The associated project identifier.
        created_by: The user who created the collaborator entry.
        updated_by: The user who last updated the collaborator entry.
        created_at: The timestamp when the collaborator entry was created.
        updated_at: The timestamp when the collaborator entry was last updated.
    """

    collaborator: str
    role: ProjectCollaboratorRole
    project_id: str
    created_by: str
    updated_by: str
    created_at: datetime.datetime
    updated_at: datetime.datetime

FileTransferStatus

Bases: str, Enum

Represents the status of a file transfer.

Attributes:

Name	Type	Description
PENDING		The transfer is pending.
SUCCESS		The transfer was successful.
FAILED		The transfer failed.

Source code in qfieldcloud_sdk/sdk.py

class FileTransferStatus(str, Enum):
    """Represents the status of a file transfer.

    Attributes:
        PENDING: The transfer is pending.
        SUCCESS: The transfer was successful.
        FAILED: The transfer failed.
    """

    PENDING = "PENDING"
    SUCCESS = "SUCCESS"
    FAILED = "FAILED"

FileTransferType

Bases: str, Enum

Represents the type of file transfer.

The PACKAGE transfer type is used only internally in QFieldCloud workers, so it should never be used by other API clients.

Attributes:

Name	Type	Description
PROJECT		Refers to a project file.
PACKAGE		Refers to a package Type.

Source code in qfieldcloud_sdk/sdk.py

class FileTransferType(str, Enum):
    """Represents the type of file transfer.

    The PACKAGE transfer type is used only internally in QFieldCloud workers, so it should never be used by other API clients.

    Attributes:
        PROJECT: Refers to a project file.
        PACKAGE: Refers to a package Type.
    """

    PROJECT = "project"
    PACKAGE = "package"

JobTypes

Bases: str, Enum

Represents the types of jobs that can be processed on QFieldCloud.

Attributes:

Name	Type	Description
PACKAGE		Refers to a packaging job.
APPLY_DELTAS		Refers to applying deltas (differences).
PROCESS_PROJECTFILE		Refers to processing a project file.

Source code in qfieldcloud_sdk/sdk.py

class JobTypes(str, Enum):
    """Represents the types of jobs that can be processed on QFieldCloud.

    Attributes:
        PACKAGE: Refers to a packaging job.
        APPLY_DELTAS: Refers to applying deltas (differences).
        PROCESS_PROJECTFILE: Refers to processing a project file.
    """

    PACKAGE = "package"
    APPLY_DELTAS = "delta_apply"
    PROCESS_PROJECTFILE = "process_projectfile"

OrganizationMemberModel

Bases: TypedDict

Represents the structure of an organization member in the QFieldCloud system.

Attributes:

Name	Type	Description
member	str	The member's identifier.
role	OrganizationMemberRole	The role of the member.
organization	str	The associated organization identifier.
is_public	bool	A boolean indicating if the membership is public.

Source code in qfieldcloud_sdk/sdk.py

class OrganizationMemberModel(TypedDict):
    """Represents the structure of an organization member in the QFieldCloud system.

    Attributes:
        member: The member's identifier.
        role: The role of the member.
        organization: The associated organization identifier.
        is_public: A boolean indicating if the membership is public.
    """

    member: str
    role: OrganizationMemberRole
    organization: str
    is_public: bool

OrganizationMemberRole

Bases: str, Enum

Defines roles for organization members.

See organization member roles documentation: https://docs.qfield.org/reference/qfieldcloud/permissions/#roles_2

Attributes:

Name	Type	Description
ADMIN		Administrator role.
MEMBER		Member role.

Source code in qfieldcloud_sdk/sdk.py

class OrganizationMemberRole(str, Enum):
    """Defines roles for organization members.

    See organization member roles documentation: https://docs.qfield.org/reference/qfieldcloud/permissions/#roles_2

    Attributes:
        ADMIN: Administrator role.
        MEMBER: Member role.
    """

    ADMIN = "admin"
    MEMBER = "member"

Pagination

The Pagination class allows for controlling and managing pagination of results within the QFieldCloud SDK.

Parameters:

Name	Type	Description	Default
limit	Optional[int]	The maximum number of items to return. Defaults to None.	None
offset	Optional[int]	The starting point from which to return items. Defaults to None.	None

Attributes:

Name	Type	Description
limit	Optional[int]	The maximum number of items to return.
offset	Optional[int]	The starting point from which to return items.

Source code in qfieldcloud_sdk/sdk.py

class Pagination:
    """The Pagination class allows for controlling and managing pagination of results within the QFieldCloud SDK.

    Args:
        limit: The maximum number of items to return. Defaults to None.
        offset: The starting point from which to return items. Defaults to None.

    Attributes:
        limit: The maximum number of items to return.
        offset: The starting point from which to return items.
    """

    limit: Optional[int] = None
    offset: Optional[int] = None

    def __init__(
        self, limit: Optional[int] = None, offset: Optional[int] = None
    ) -> None:
        self.limit = limit
        self.offset = offset

    @property
    def is_empty(self) -> bool:
        """Whether both limit and offset are None, indicating no pagination settings.

        Returns:
            True if both limit and offset are None, False otherwise.
        """
        return self.limit is None and self.offset is None

is_empty: bool property

Whether both limit and offset are None, indicating no pagination settings.

Returns:

Type	Description
bool	True if both limit and offset are None, False otherwise.

ProjectCollaboratorRole

Bases: str, Enum

Defines roles for project collaborators.

See project collaborator roles documentation: https://docs.qfield.org/reference/qfieldcloud/permissions/#roles_1

Attributes:

Name	Type	Description
ADMIN		Administrator role.
MANAGER		Manager role.
EDITOR		Editor role.
REPORTER		Reporter role.
READER		Reader role.

Source code in qfieldcloud_sdk/sdk.py

class ProjectCollaboratorRole(str, Enum):
    """Defines roles for project collaborators.

    See project collaborator roles documentation: https://docs.qfield.org/reference/qfieldcloud/permissions/#roles_1

    Attributes:
        ADMIN: Administrator role.
        MANAGER: Manager role.
        EDITOR: Editor role.
        REPORTER: Reporter role.
        READER: Reader role.
    """

    ADMIN = "admin"
    MANAGER = "manager"
    EDITOR = "editor"
    REPORTER = "reporter"
    READER = "reader"