Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
145 changes: 144 additions & 1 deletion qfieldcloud_sdk/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -268,7 +268,7 @@ def delete_project(ctx, project_id):
# print_json(payload)
print(payload, payload.content)
else:
log(f'Delеted project "{project_id}".')
log(f'Deleted project "{project_id}".')


@cli.command()
Expand Down Expand Up @@ -529,3 +529,146 @@ def package_download(
)
else:
log(f"No packaged files to download for project {project_id}")


@cli.command(short_help="Get a list of project collaborators.")
@click.argument("project_id")
@click.pass_context
def collaborators_get(ctx, project_id: str) -> None:
"""Get a list of project collaborators for specific project with PROJECT_ID."""
collaborators = ctx.obj["client"].get_project_collaborators(project_id)

if ctx.obj["format_json"]:
print_json(collaborators)
else:
log(f'Collaborators for project with id "{project_id}":')
for collaborator in collaborators:
log(f'{collaborator["collaborator"]}\t{collaborator["role"]}')


@cli.command(short_help="Add a project collaborator.")
@click.argument("project_id")
@click.argument("username")
@click.argument("role", type=sdk.ProjectCollaboratorRole)
@click.pass_context
def collaborators_add(
ctx, project_id: str, username: str, role: sdk.ProjectCollaboratorRole
) -> None:
"""Add collaborator with USERNAME with specific ROLE to a project with PROJECT_ID. Possible ROLE values: admin, manager, editor, reporter, reader."""
collaborator = ctx.obj["client"].add_project_collaborator(
project_id, username, role
)

if ctx.obj["format_json"]:
print_json(collaborator)
else:
log(
f'Collaborator "{collaborator["collaborator"]}" added to project with id "{collaborator["project_id"]}" with role "{collaborator["role"]}".'
)


@cli.command(short_help="Remove a project collaborator.")
@click.argument("project_id")
@click.argument("username")
@click.pass_context
def collaborators_remove(ctx, project_id: str, username: str) -> None:
"""Remove collaborator with USERNAME from project with PROJECT_ID."""
ctx.obj["client"].remove_project_collaborators(project_id, username)

if not ctx.obj["format_json"]:
log(f'Collaborator "{username}" removed project with id "{project_id}".')


@cli.command(short_help="Change project collaborator role.")
@click.argument("project_id")
@click.argument("username")
@click.argument("role", type=sdk.ProjectCollaboratorRole)
@click.pass_context
def collaborators_patch(
ctx, project_id: str, username: str, role: sdk.ProjectCollaboratorRole
) -> None:
"""Change collaborator with USERNAME to new ROLE in project with PROJECT_ID. Possible ROLE values: admin, manager, editor, reporter, reader."""
collaborator = ctx.obj["client"].patch_project_collaborators(
project_id, username, role
)

if ctx.obj["format_json"]:
print_json(collaborator)
else:
log(
f'Collaborator "{collaborator["collaborator"]}" added to project with id "{collaborator["project_id"]}" with role "{collaborator["role"]}".'
)


@cli.command(short_help="Get a list organization members.")
@click.argument("organization")
@click.pass_context
def members_get(ctx, organization: str) -> None:
"""Get a list of ORGANIZATION members."""
memberships = ctx.obj["client"].get_organization_members(organization)

if ctx.obj["format_json"]:
print_json(memberships)
else:
log(f'Members of organization "{organization}":')
for membership in memberships:
log(f'{membership["member"]}\t{membership["role"]}')


@cli.command(short_help="Add an organization member.")
@click.argument("organization")
@click.argument("username")
@click.argument("role", type=sdk.OrganizationMemberRole)
@click.option("--public/--no-public", "is_public")
@click.pass_context
def members_add(
ctx,
organization: str,
username: str,
role: sdk.OrganizationMemberRole,
is_public: bool,
) -> None:
"""Add member with USERNAME with ROLE to ORGANIZATION. Possible ROLE values: admin, member."""
membership = ctx.obj["client"].add_organization_member(
organization, username, role, is_public
)

if ctx.obj["format_json"]:
print_json(membership)
else:
log(
f'Member "{membership["member"]}" added to organization "{membership["organization"]}" with role "{membership["role"]}".'
)


@cli.command(short_help="Remove an organization member.")
@click.argument("organization")
@click.argument("username")
@click.pass_context
def members_remove(ctx, organization: str, username: str) -> None:
"""Remove member with USERNAME from ORGANIZATION."""
ctx.obj["client"].remove_organization_members(organization, username)

if not ctx.obj["format_json"]:
log(f'Member "{username}" removed organization "{organization}".')


@cli.command(short_help="Change organization member role.")
@click.argument("organization")
@click.argument("username")
@click.argument("role", type=sdk.OrganizationMemberRole)
@click.pass_context
def members_patch(
ctx, organization: str, username: str, role: sdk.OrganizationMemberRole
) -> None:
"""Change member with USERNAME to new ROLE in ORGANIZATION. Possible ROLE values: admin, member."""
membership = ctx.obj["client"].patch_organization_members(
organization, username, role
)

if ctx.obj["format_json"]:
print_json(membership)
else:
log(
f'Member "{membership["member"]}" changed role in organization "{membership["organization"]}" to role "{membership["role"]}".'
)
201 changes: 200 additions & 1 deletion qfieldcloud_sdk/sdk.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
import datetime
import fnmatch
import logging
import os
import sys
from enum import Enum
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Union, cast
from typing import Any, Callable, Dict, List, Optional, TypedDict, Union, cast
from urllib import parse as urlparse

import requests
Expand Down Expand Up @@ -48,6 +49,41 @@ class JobTypes(str, Enum):
PROCESS_PROJECTFILE = "process_projectfile"


class ProjectCollaboratorRole(str, Enum):
ADMIN = "admin"
MANAGER = "manager"
EDITOR = "editor"
REPORTER = "reporter"
READER = "reader"


class OrganizationMemberRole(str, Enum):
ADMIN = "admin"
MEMBER = "member"


class CollaboratorModel(TypedDict):
collaborator: str
role: ProjectCollaboratorRole
project_id: str
created_by: str
updated_by: str
created_at: datetime.datetime
updated_at: datetime.datetime


class OrganizationMemberModel(TypedDict):
member: str
role: OrganizationMemberRole
organization: str
is_public: bool
# TODO future work that can be surely expected, check QF-4535
# created_by: str
# updated_by: str
# created_at: datetime.datetime
# updated_at: datetime.datetime


class Pagination:
limit = None
offset = None
Expand Down Expand Up @@ -680,6 +716,169 @@ def list_local_files(

return files

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

Args:
project_id (str): project UUID

Returns:
List[CollaboratorModel]: the list of collaborators for that project
"""
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 (str): project UUID
username (str): username of the collaborator to be added
role (ProjectCollaboratorRole): the role of the collaborator. One of: `reader`, `reporter`, `editor`, `manager` or `admin`

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

return collaborator

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

Args:
project_id (str): project UUID
username (str): 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 collaborator

Args:
project_id (str): project UUID
username (str): the username of the collaborator to be patched
role (ProjectCollaboratorRole): the new role of the collaborator

Returns:
CollaboratorModel: the updated 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 project members.

Args:
organization (str): organization username

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

return members

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

Args:
organization (str): organization username
username (str): username of the member to be added
role (OrganizationMemberRole): the role of the member. One of: `admin` or `member`.

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

return member

def remove_organization_members(self, project_id: str, username: str) -> None:
"""Removes a member from a project.

Args:
project_id (str): project UUID
username (str): 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 member

Args:
project_id (str): project UUID
username (str): the username of the member to be patched
role (OrganizationMemberRole): the new role of the member

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

return member

def _request_json(
self,
method: str,
Expand Down