NextCloud.Group.get_subadmins() - Code Metrics - Inspection of "Merge pull request #15 from danil-topc/activities_..." - EnterpriseyIntranet/nextcloud-API - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 52addc...0b5289 )

by Matěj

created 2019-01-14 21:14 UTC

NextCloud.Group.get_subadmins() A

↳ Parent: NextCloud

Complexity

Conditions

Size

Total Lines	9
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	9
rs	10
c	0
b	0
f	0
cc	1
nop	2

import enum
import requests

PUBLIC_API_NAME_CLASS_MAP = dict()


class Requester(object):
    def __init__(self, endpoint, user, passwd, js=False):
        self.query_components = []

        self.to_json = js

        self.base_url = endpoint
        # GroupFolders.url = endpoint + "/ocs/v2.php/apps/groupfolders/folders"

        self.h_get = {"OCS-APIRequest": "true"}
        self.h_post = {"OCS-APIRequest": "true",
                       "Content-Type": "application/x-www-form-urlencoded"}
        self.auth_pk = (user, passwd)
        self.API_URL = None

    def rtn(self, resp):
        if self.to_json:
            return resp.json()
        else:
            return resp.content.decode("UTF-8")

    def get(self, url="", params=None):
        url = self.get_full_url(url)
        res = requests.get(url, auth=self.auth_pk, headers=self.h_get, params=params)
        return self.rtn(res)

    def post(self, url="", data=None):
        url = self.get_full_url(url)
        res = requests.post(url, auth=self.auth_pk, data=data, headers=self.h_post)
        return self.rtn(res)

    def put(self, url="", data=None):
        url = self.get_full_url(url)
        res = requests.put(url, auth=self.auth_pk, data=data, headers=self.h_post)
        return self.rtn(res)

    def delete(self, url="", data=None):
        url = self.get_full_url(url)
        res = requests.delete(url, auth=self.auth_pk, data=data, headers=self.h_post)
        return self.rtn(res)

    def get_full_url(self, additional_url=""):
        """
        Build full url for request to NextCloud api

        Construct url from self.base_url, self.API_URL, additional_url (if given), add format=json param if self.json

        :param additional_url: str
            add to url after api_url
        :return: str
        """
        if additional_url and not str(additional_url).startswith("/"):
            additional_url = "/{}".format(additional_url)

        if self.to_json:
            self.query_components.append("format=json")

        ret = "{base_url}{api_url}{additional_url}".format(
            base_url=self.base_url, api_url=self.API_URL, additional_url=additional_url)

        if self.to_json:
            ret += "?format=json"
        return ret


def nextcloud_method(method_to_wrap):
    class_name = method_to_wrap.__qualname__.split(".", 1)[0]
    PUBLIC_API_NAME_CLASS_MAP[method_to_wrap.__name__] = class_name
    return method_to_wrap


class NextCloud(object):

    def __init__(self, endpoint, user, password, js=False):
        self.query_components = []

        requester = Requester(endpoint, user, password, js)

        self.functionality = {
            "Apps": Apps(requester),
            "Group": Group(requester),
            "GroupFolders": GroupFolders(requester),
            "Share": Share(requester),
            "User": User(requester),
            "FederatedCloudShare": FederatedCloudShare(requester),
            "Activity": Activity(requester),
        }
        for name, location in PUBLIC_API_NAME_CLASS_MAP.items():
            setattr(self, name, getattr(self.functionality[location], name))


class WithRequester(object):

    API_URL = NotImplementedError

    def __init__(self, requester):
        self._requester = requester

    @property
    def requester(self):
        """ Get requester instance """
        # dynamically set API_URL for requester
        self._requester.API_URL = self.API_URL
        return self._requester


class GroupFolders(WithRequester):
    API_URL = "/apps/groupfolders/folders"

    @nextcloud_method
    def get_group_folders(self):
        """
        Return a list of call configured folders and their settings

        Returns:

        """
        return self.requester.get()

    @nextcloud_method
    def get_group_folder(self, fid):
        """
        Return a specific configured folder and it's settings

        Args:
            fid (int/str): group folder id

        Returns:

        """
        return self.requester.get(fid)

    @nextcloud_method
    def create_group_folder(self, mountpoint):
        """
        Create a new group folder

        Args:
            mountpoint (str): name for the new folder

        Returns:

        """
        return self.requester.post(data={"mountpoint": mountpoint})

    @nextcloud_method
    def delete_group_folder(self, fid):
        """
        Delete a group folder

        Args:
            fid (int/str): group folder id

        Returns:

        """
        return self.requester.delete(fid)

    @nextcloud_method
    def grant_access_to_group_folder(self, fid, gid):
        """
        Give a group access to a folder

        Args:
            fid (int/str): group folder id
            gid (str): group to share with id

        Returns:

        """
        url = "/".join([str(fid), "groups"])
        return self.requester.post(url, data={"group": gid})

    @nextcloud_method
    def revoke_access_to_group_folder(self, fid, gid):
        """
        Remove access from a group to a folder

        Args:
            fid (int/str): group folder id
            gid (str): group id

        Returns:

        """
        url = "/".join([str(fid), "groups", gid])
        return self.requester.delete(url)

    @nextcloud_method
    def set_permissions_to_group_folder(self, fid, gid, permissions):
        """
        Set the permissions a group has in a folder

        Args:
            fid (int/str): group folder id
            gid (str): group id
            permissions (int): The new permissions for the group as attribute of Permission class

        Returns:

        """
        url = "/".join([str(fid), "groups", gid])
        return self.requester.post(url=url, data={"permissions": permissions})

    @nextcloud_method
    def set_quota_of_group_folder(self, fid, quota):
        """
        Set the quota for a folder in bytes

        Args:
            fid (int/str): group folder id
            quota (int/str): The new quota for the folder in bytes, user -3 for unlimited

        Returns:

        """
        url = "/".join([str(fid), "quota"])
        return self.requester.post(url, {"quota": quota})

    @nextcloud_method
    def rename_group_folder(self, fid, mountpoint):
        """
        Change the name of a folder

        Args:
            fid (int/str): group folder id
            mountpoint (str): The new name for the folder

        Returns:

        """
        url = "/".join([str(fid), "mountpoint"])
        return self.requester.post(url=url, data={"mountpoint": mountpoint})


class Share(WithRequester):
    API_URL = "/ocs/v2.php/apps/files_sharing/api/v1"
    LOCAL = "shares"

    def get_local_url(self, additional_url=""):
        if additional_url:
            return "/".join([self.LOCAL, additional_url])
        return self.LOCAL

    @staticmethod
    def validate_share_parameters(path, share_type, share_with):
        """
        Check if share parameters make sense

        Args:
            path (str): path to the file/folder which should be shared
            share_type (int): ShareType attribute
            share_with (str): user/group id with which the file should be shared

        Returns:
            bool: True if parameters make sense together, False otherwise
        """
        if (path is None or not isinstance(share_type, int)) \
                or (share_type in [ShareType.GROUP, ShareType.USER, ShareType.FEDERATED_CLOUD_SHARE]
                    and share_with is None):
            return False
        return True

    @nextcloud_method
    def get_shares(self):
        """ Get all shares from the user """
        return self.requester.get(self.get_local_url())

    @nextcloud_method
    def get_shares_from_path(self, path, reshares=None, subfiles=None):
        """
        Get all shares from a given file/folder

        Args:
            path (str): path to file/folder
            reshares (bool): (optional) return not only the shares from the current user but all shares from the given file
            subfiles (bool): (optional) return all shares within a folder, given that path defines a folder

        Returns:

        """
        url = self.get_local_url()
        params = {
            "path": path,
            "reshares": None if reshares is None else str(bool(reshares)).lower(),  # TODO: test reshares, subfiles
            "subfiles": None if subfiles is None else str(bool(subfiles)).lower(),
        }
        return self.requester.get(url, params=params)

    @nextcloud_method
    def get_share_info(self, sid):
        """
        Get information about a given share

        Args:
            sid (int): share id

        Returns:
        """
        return self.requester.get(self.get_local_url(sid))

    @nextcloud_method
    def create_share(
            self, path, share_type, share_with=None, public_upload=None,
            password=None, permissions=None):
        """
        Share a file/folder with a user/group or as public link

        Mandatory fields: share_type, path and share_with for share_type USER (0) or GROUP (1).

        Args:
            path (str): path to the file/folder which should be shared
            share_type (int): ShareType attribute
            share_with (str): user/group id with which the file should be shared
            public_upload (bool): bool, allow public upload to a public shared folder (true/false)
            password (str): password to protect public link Share with
            permissions (int): sum of selected Permission attributes

        Returns:

        """
        if not self.validate_share_parameters(path, share_type, share_with):
            return False

        url = self.get_local_url()
        if public_upload:
            public_upload = "true"

        data = {"path": path, "shareType": share_type}
        if share_type in [ShareType.GROUP, ShareType.USER, ShareType.FEDERATED_CLOUD_SHARE]:
            data["shareWith"] = share_with
        if public_upload:
            data["publicUpload"] = public_upload
        if share_type == ShareType.PUBLIC_LINK and password is not None:
            data["password"] = str(password)
        if permissions is not None:
            data["permissions"] = permissions
        return self.requester.post(url, data)

    @nextcloud_method
    def delete_share(self, sid):
        """
        Remove the given share

        Args:
            sid (str): share id

        Returns:

        """
        return self.requester.delete(self.get_local_url(sid))

    @nextcloud_method
    def update_share(self, sid, permissions=None, password=None, public_upload=None, expire_date=""):
        """
        Update a given share, only one value can be updated per request

        Args:
            sid (str): share id
            permissions (int): sum of selected Permission attributes
            password (str): password to protect public link Share with
            public_upload (bool): bool, allow public upload to a public shared folder (true/false)
            expire_date (str): set an expire date for public link shares. Format: ‘YYYY-MM-DD’

        Returns:

        """
        params = dict(
            permissions=permissions,
            password=password,
            expireDate=expire_date
        )
        if public_upload:
            params["publicUpload"] = "true"
        if public_upload is False:
            params["publicUpload"] = "false"

        # check if only one param specified
        specified_params_count = sum([int(bool(each)) for each in params.values()])
        if specified_params_count > 1:
            raise ValueError("Only one parameter for update can be specified per request")

        url = self.get_local_url(sid)
        return self.requester.put(url, data=params)


class FederatedCloudShare(WithRequester):
    API_URL = "/ocs/v2.php/apps/files_sharing/api/v1"
    FEDERATED = "remote_shares"

    def get_federated_url(self, additional_url=""):
        if additional_url:
            return "/".join([self.FEDERATED, additional_url])
        return self.FEDERATED

    @nextcloud_method
    def list_accepted_federated_cloudshares(self):
        url = self.get_federated_url()
        return self.requester.get(url)

    @nextcloud_method
    def get_known_federated_cloudshare(self, sid):
        url = self.get_federated_url(sid)
        return self.requester.get(url)

    @nextcloud_method
    def delete_accepted_federated_cloudshare(self, sid):
        url = self.get_federated_url(sid)
        return self.requester.delete(url)

    @nextcloud_method
    def list_pending_federated_cloudshares(self, sid):
        url = self.get_federated_url("pending")
        return self.requester.get(url)

    @nextcloud_method
    def accept_pending_federated_cloudshare(self, sid):
        url = self.get_federated_url("pending/{sid}".format(sid=sid))
        return self.requester.post(url)

    @nextcloud_method
    def decline_pending_federated_cloudshare(self, sid):
        url = self.get_federated_url("pending/{sid}".format(sid=sid))
        return self.requester.delete(url)


class Apps(WithRequester):
    API_URL = "/ocs/v1.php/cloud/apps"

    @nextcloud_method
    def get_apps(self, filter=None):
        """
        Get a list of apps installed on the Nextcloud server

        :param filter: str, optional "enabled" or "disabled"
        :return:
        """
        params = {
            "filter": filter
        }
        return self.requester.get(params=params)

    @nextcloud_method
    def get_app(self, app_id):
        """
        Provide information on a specific application

        :param app_id: str, app id
        :return:
        """
        return self.requester.get(app_id)

    @nextcloud_method
    def enable_app(self, app_id):
        """
        Enable an app

        :param app_id: str, app id
        :return:
        """
        return self.requester.post(app_id)

    @nextcloud_method
    def disable_app(self, app_id):
        """
        Disable the specified app

        :param app_id: str, app id
        :return:
        """
        return self.requester.delete(app_id)


class Group(WithRequester):
    API_URL = "/ocs/v1.php/cloud/groups"

    @nextcloud_method
    def get_groups(self, search=None, limit=None, offset=None):
        """
        Retrieve a list of groups from the Nextcloud server

        :param search: string, optional search string
        :param limit: int, optional limit value
        :param offset: int, optional offset value
        :return:
        """
        params = {
            'search': search,
            'limit': limit,
            'offset': offset
        }
        return self.requester.get(params=params)

    @nextcloud_method
    def add_group(self, gid):
        """
        Add a new group

        :param gid: str, group name
        :return:
        """
        msg = {"groupid": gid}
        return self.requester.post("", msg)

    @nextcloud_method
    def get_group(self, gid):
        """
        Retrieve a list of group members

        :param gid: str, group name
        :return:
        """
        return self.requester.get("{gid}".format(gid=gid))

    @nextcloud_method
    def get_subadmins(self, gid):
        """
        List subadmins of the group

        :param gid: str, group name
        :return:
        """
        return self.requester.get("{gid}/subadmins".format(gid=gid))

    @nextcloud_method
    def delete_group(self, gid):
        """
        Remove a group

        :param gid: str, group name
        :return:
        """
        return self.requester.delete("{gid}".format(gid=gid))


class User(WithRequester):
    API_URL = "/ocs/v1.php/cloud/users"

    @nextcloud_method
    def add_user(self, uid, passwd):
        """
        Create a new user on the Nextcloud server

        :param uid: str, uid of new user
        :param passwd: str, password of new user
        :return:
        """
        msg = {'userid': uid, 'password': passwd}
        return self.requester.post("", msg)

    @nextcloud_method
    def get_users(self, search=None, limit=None, offset=None):
        """
        Retrieve a list of users from the Nextcloud server

        :param search: string, optional search string
        :param limit: int, optional limit value
        :param offset: int, optional offset value
        :return:
        """
        params = {
            'search': search,
            'limit': limit,
            'offset': offset
        }
        return self.requester.get(params=params)

    @nextcloud_method
    def get_user(self, uid):
        """
        Retrieve information about a single user

        :param uid: str, uid of user
        :return:
        """
        return self.requester.get("{uid}".format(uid=uid))

    @nextcloud_method
    def edit_user(self, uid, what, value):
        """
        Edit attributes related to a user

        Users are able to edit email, displayname and password; admins can also edit the quota value

        :param uid: str, uid of user
        :param what: str, the field to edit
        :param value: str, the new value for the field
        :return:
        """
        what_to_key_map = dict(
            email="email", quota="quote", phone="phone", address="address", website="website",
            twitter="twitter", displayname="displayname", password="password",
        )
        assert what in what_to_key_map, (
            "You have chosen to edit user's '{what}', but you can choose only from: {choices}"
            .format(what=what, choices=", ".join(what_to_key_map.keys()))
        )

        url = "{uid}".format(uid=uid)
        msg = dict(
            key=what_to_key_map[what],
            value=value,
        )
        return self.requester.put(url, msg)

    @nextcloud_method
    def disable_user(self, uid):
        """
        Disable a user on the Nextcloud server so that the user cannot login anymore

        :param uid: str, uid of user
        :return:
        """
        return self.requester.put("{uid}/disable".format(uid=uid))

    @nextcloud_method
    def enable_user(self, uid):
        """
        Enable a user on the Nextcloud server so that the user can login again

        :param uid: str, uid of user
        :return:
        """
        return self.requester.put("{uid}/enable".format(uid=uid))

    @nextcloud_method
    def delete_user(self, uid):
        """
        Delete a user from the Nextcloud server

        :param uid: str, uid of user
        :return:
        """
        return self.requester.delete("{uid}".format(uid=uid))

    @nextcloud_method
    def add_to_group(self, uid, gid):
        """
        Add the specified user to the specified group

        :param uid: str, uid of user
        :param gid: str, name of group
        :return:
        """
        url = "{uid}/groups".format(uid=uid)
        msg = {'groupid': gid}
        return self.requester.post(url, msg)

    @nextcloud_method
    def remove_from_group(self, uid, gid):
        """
        Remove the specified user from the specified group

        :param uid: str, uid of user
        :param gid: str, name of group
        :return:
        """
        url = "{uid}/groups".format(uid=uid)
        msg = {'groupid': gid}
        return self.requester.delete(url, msg)

    @nextcloud_method
    def create_subadmin(self, uid, gid):
        """
        Make a user the subadmin of a group

        :param uid: str, uid of user
        :param gid: str, name of group
        :return:
        """
        url = "{uid}/subadmins".format(uid=uid)
        msg = {'groupid': gid}
        return self.requester.post(url, msg)

    @nextcloud_method
    def remove_subadmin(self, uid, gid):
        """
        Remove the subadmin rights for the user specified from the group specified

        :param uid: str, uid of user
        :param gid: str, name of group
        :return:
        """
        url = "{uid}/subadmins".format(uid=uid)
        msg = {'groupid': gid}
        return self.requester.delete(url, msg)

    @nextcloud_method
    def get_subadmin_groups(self, uid):
        """
        Get the groups in which the user is a subadmin

        :param uid: str, uid of user
        :return:
        """
        url = "{uid}/subadmins".format(uid=uid)
        return self.requester.get(url)

    @nextcloud_method
    def resend_welcome_mail(self, uid):
        """
        Trigger the welcome email for this user again

        :param uid: str, uid of user
        :return:
        """
        url = "{uid}/welcome".format(uid=uid)
        return self.requester.post(url)


class Activity(WithRequester):
    API_URL = "/ocs/v2.php/apps/activity/api/v2/activity"

    @nextcloud_method
    def get_activities(self, since=None, limit=None, object_type=None, object_id=None, sort=None):
        """
        Get an activity feed showing your file changes and other interesting things going on in your Nextcloud

        All args are optional

        Args:
            since (int): ID of the last activity that you’ve seen
            limit (int): How many activities should be returned (default: 50)
            object_type (string): Filter the activities to a given object. May only appear together with object_id
            object_id (string): Filter the activities to a given object. May only appear together with object_type
            sort (str, "asc" or "desc"): Sort activities ascending or descending (from the since) (Default: desc)

        Returns:

        """
        params = dict(
            since=since,
            limit=limit,
            object_type=object_type,
            object_id=object_id,
            sort=sort
        )
        if params['object_type'] and params['object_id']:
            return self.requester.get(url="filter", params=params)
        return self.requester.get(params=params)


class OCSCode(enum.IntEnum):
    OK = 100
    SERVER_ERROR = 996
    NOT_AUTHORIZED = 997
    NOT_FOUND = 998
    UNKNOWN_ERROR = 999


class ShareType(enum.IntEnum):
    USER = 0
    GROUP = 1
    PUBLIC_LINK = 3
    FEDERATED_CLOUD_SHARE = 6


class Permission(enum.IntEnum):
    """ Permission for Share have to be sum of selected permissions """
    READ = 1
    UPDATE = 2
    CREATE = 4
    DELETE = 8
    SHARE = 16
    ALL = 31


QUOTA_UNLIMITED = -3


def datetime_to_expire_date(date):
    return date.strftime("%Y-%m-%d")


1			import enum
2			import requests
3
4			PUBLIC_API_NAME_CLASS_MAP = dict()
5
6
7			class Requester(object):
8			def __init__(self, endpoint, user, passwd, js=False):
9			self.query_components = []
10
11			self.to_json = js
12
13			self.base_url = endpoint
14			# GroupFolders.url = endpoint + "/ocs/v2.php/apps/groupfolders/folders"
15
16			self.h_get = {"OCS-APIRequest": "true"}
17			self.h_post = {"OCS-APIRequest": "true",
18			"Content-Type": "application/x-www-form-urlencoded"}
19			self.auth_pk = (user, passwd)
20			self.API_URL = None
21
22			def rtn(self, resp):
23			if self.to_json:
24			return resp.json()
25			else:
26			return resp.content.decode("UTF-8")
27
28			def get(self, url="", params=None):
29			url = self.get_full_url(url)
30			res = requests.get(url, auth=self.auth_pk, headers=self.h_get, params=params)
31			return self.rtn(res)
32
33			def post(self, url="", data=None):
34			url = self.get_full_url(url)
35			res = requests.post(url, auth=self.auth_pk, data=data, headers=self.h_post)
36			return self.rtn(res)
37
38			def put(self, url="", data=None):
39			url = self.get_full_url(url)
40			res = requests.put(url, auth=self.auth_pk, data=data, headers=self.h_post)
41			return self.rtn(res)
42
43			def delete(self, url="", data=None):
44			url = self.get_full_url(url)
45			res = requests.delete(url, auth=self.auth_pk, data=data, headers=self.h_post)
46			return self.rtn(res)
47
48			def get_full_url(self, additional_url=""):
49			"""
50			Build full url for request to NextCloud api
51
52			Construct url from self.base_url, self.API_URL, additional_url (if given), add format=json param if self.json
53
54			:param additional_url: str
55			add to url after api_url
56			:return: str
57			"""
58			if additional_url and not str(additional_url).startswith("/"):
59			additional_url = "/{}".format(additional_url)
60
61			if self.to_json:
62			self.query_components.append("format=json")
63
64			ret = "{base_url}{api_url}{additional_url}".format(
65			base_url=self.base_url, api_url=self.API_URL, additional_url=additional_url)
66
67			if self.to_json:
68			ret += "?format=json"
69			return ret
70
71
72			def nextcloud_method(method_to_wrap):
73			class_name = method_to_wrap.__qualname__.split(".", 1)[0]
74			PUBLIC_API_NAME_CLASS_MAP[method_to_wrap.__name__] = class_name
75			return method_to_wrap
76
77
78			class NextCloud(object):
79
80			def __init__(self, endpoint, user, password, js=False):
81			self.query_components = []
82
83			requester = Requester(endpoint, user, password, js)
84
85			self.functionality = {
86			"Apps": Apps(requester),
87			"Group": Group(requester),
88			"GroupFolders": GroupFolders(requester),
89			"Share": Share(requester),
90			"User": User(requester),
91			"FederatedCloudShare": FederatedCloudShare(requester),
92			"Activity": Activity(requester),
93			}
94			for name, location in PUBLIC_API_NAME_CLASS_MAP.items():
95			setattr(self, name, getattr(self.functionality[location], name))
96
97
98			class WithRequester(object):
99
100			API_URL = NotImplementedError
101
102			def __init__(self, requester):
103			self._requester = requester
104
105			@property
106			def requester(self):
107			""" Get requester instance """
108			# dynamically set API_URL for requester
109			self._requester.API_URL = self.API_URL
110			return self._requester
111
112
113			class GroupFolders(WithRequester):
114			API_URL = "/apps/groupfolders/folders"
115
116			@nextcloud_method
117			def get_group_folders(self):
118			"""
119			Return a list of call configured folders and their settings
120
121			Returns:
122
123			"""
124			return self.requester.get()
125
126			@nextcloud_method
127			def get_group_folder(self, fid):
128			"""
129			Return a specific configured folder and it's settings
130
131			Args:
132			fid (int/str): group folder id
133
134			Returns:
135
136			"""
137			return self.requester.get(fid)
138
139			@nextcloud_method
140			def create_group_folder(self, mountpoint):
141			"""
142			Create a new group folder
143
144			Args:
145			mountpoint (str): name for the new folder
146
147			Returns:
148
149			"""
150			return self.requester.post(data={"mountpoint": mountpoint})
151
152			@nextcloud_method
153			def delete_group_folder(self, fid):
154			"""
155			Delete a group folder
156
157			Args:
158			fid (int/str): group folder id
159
160			Returns:
161
162			"""
163			return self.requester.delete(fid)
164
165			@nextcloud_method
166			def grant_access_to_group_folder(self, fid, gid):
167			"""
168			Give a group access to a folder
169
170			Args:
171			fid (int/str): group folder id
172			gid (str): group to share with id
173
174			Returns:
175
176			"""
177			url = "/".join([str(fid), "groups"])
178			return self.requester.post(url, data={"group": gid})
179
180			@nextcloud_method
181			def revoke_access_to_group_folder(self, fid, gid):
182			"""
183			Remove access from a group to a folder
184
185			Args:
186			fid (int/str): group folder id
187			gid (str): group id
188
189			Returns:
190
191			"""
192			url = "/".join([str(fid), "groups", gid])
193			return self.requester.delete(url)
194
195			@nextcloud_method
196			def set_permissions_to_group_folder(self, fid, gid, permissions):
197			"""
198			Set the permissions a group has in a folder
199
200			Args:
201			fid (int/str): group folder id
202			gid (str): group id
203			permissions (int): The new permissions for the group as attribute of Permission class
204
205			Returns:
206
207			"""
208			url = "/".join([str(fid), "groups", gid])
209			return self.requester.post(url=url, data={"permissions": permissions})
210
211			@nextcloud_method
212			def set_quota_of_group_folder(self, fid, quota):
213			"""
214			Set the quota for a folder in bytes
215
216			Args:
217			fid (int/str): group folder id
218			quota (int/str): The new quota for the folder in bytes, user -3 for unlimited
219
220			Returns:
221
222			"""
223			url = "/".join([str(fid), "quota"])
224			return self.requester.post(url, {"quota": quota})
225
226			@nextcloud_method
227			def rename_group_folder(self, fid, mountpoint):
228			"""
229			Change the name of a folder
230
231			Args:
232			fid (int/str): group folder id
233			mountpoint (str): The new name for the folder
234
235			Returns:
236
237			"""
238			url = "/".join([str(fid), "mountpoint"])
239			return self.requester.post(url=url, data={"mountpoint": mountpoint})
240
241
242			class Share(WithRequester):
243			API_URL = "/ocs/v2.php/apps/files_sharing/api/v1"
244			LOCAL = "shares"
245
246			def get_local_url(self, additional_url=""):
247			if additional_url:
248			return "/".join([self.LOCAL, additional_url])
249			return self.LOCAL
250
251			@staticmethod
252			def validate_share_parameters(path, share_type, share_with):
253			"""
254			Check if share parameters make sense
255
256			Args:
257			path (str): path to the file/folder which should be shared
258			share_type (int): ShareType attribute
259			share_with (str): user/group id with which the file should be shared
260
261			Returns:
262			bool: True if parameters make sense together, False otherwise
263			"""
264			if (path is None or not isinstance(share_type, int)) \
265			or (share_type in [ShareType.GROUP, ShareType.USER, ShareType.FEDERATED_CLOUD_SHARE]
266			and share_with is None):
267			return False
268			return True
269
270			@nextcloud_method
271			def get_shares(self):
272			""" Get all shares from the user """
273			return self.requester.get(self.get_local_url())
274
275			@nextcloud_method
276			def get_shares_from_path(self, path, reshares=None, subfiles=None):
277			"""
278			Get all shares from a given file/folder
279
280			Args:
281			path (str): path to file/folder
282			reshares (bool): (optional) return not only the shares from the current user but all shares from the given file
283			subfiles (bool): (optional) return all shares within a folder, given that path defines a folder
284
285			Returns:
286
287			"""
288			url = self.get_local_url()
289			params = {
290			"path": path,
291			"reshares": None if reshares is None else str(bool(reshares)).lower(), # TODO: test reshares, subfiles
292			"subfiles": None if subfiles is None else str(bool(subfiles)).lower(),
293			}
294			return self.requester.get(url, params=params)
295
296			@nextcloud_method
297			def get_share_info(self, sid):
298			"""
299			Get information about a given share
300
301			Args:
302			sid (int): share id
303
304			Returns:
305			"""
306			return self.requester.get(self.get_local_url(sid))
307
308			@nextcloud_method
309			def create_share(
310			self, path, share_type, share_with=None, public_upload=None,
311			password=None, permissions=None):
312			"""
313			Share a file/folder with a user/group or as public link
314
315			Mandatory fields: share_type, path and share_with for share_type USER (0) or GROUP (1).
316
317			Args:
318			path (str): path to the file/folder which should be shared
319			share_type (int): ShareType attribute
320			share_with (str): user/group id with which the file should be shared
321			public_upload (bool): bool, allow public upload to a public shared folder (true/false)
322			password (str): password to protect public link Share with
323			permissions (int): sum of selected Permission attributes
324
325			Returns:
326
327			"""
328			if not self.validate_share_parameters(path, share_type, share_with):
329			return False
330
331			url = self.get_local_url()
332			if public_upload:
333			public_upload = "true"
334
335			data = {"path": path, "shareType": share_type}
336			if share_type in [ShareType.GROUP, ShareType.USER, ShareType.FEDERATED_CLOUD_SHARE]:
337			data["shareWith"] = share_with
338			if public_upload:
339			data["publicUpload"] = public_upload
340			if share_type == ShareType.PUBLIC_LINK and password is not None:
341			data["password"] = str(password)
342			if permissions is not None:
343			data["permissions"] = permissions
344			return self.requester.post(url, data)
345
346			@nextcloud_method
347			def delete_share(self, sid):
348			"""
349			Remove the given share
350
351			Args:
352			sid (str): share id
353
354			Returns:
355
356			"""
357			return self.requester.delete(self.get_local_url(sid))
358
359			@nextcloud_method
360			def update_share(self, sid, permissions=None, password=None, public_upload=None, expire_date=""):
361			"""
362			Update a given share, only one value can be updated per request
363
364			Args:
365			sid (str): share id
366			permissions (int): sum of selected Permission attributes
367			password (str): password to protect public link Share with
368			public_upload (bool): bool, allow public upload to a public shared folder (true/false)
369			expire_date (str): set an expire date for public link shares. Format: ‘YYYY-MM-DD’
370
371			Returns:
372
373			"""
374			params = dict(
375			permissions=permissions,
376			password=password,
377			expireDate=expire_date
378			)
379			if public_upload:
380			params["publicUpload"] = "true"
381			if public_upload is False:
382			params["publicUpload"] = "false"
383
384			# check if only one param specified
385			specified_params_count = sum([int(bool(each)) for each in params.values()])
386			if specified_params_count > 1:
387			raise ValueError("Only one parameter for update can be specified per request")
388
389			url = self.get_local_url(sid)
390			return self.requester.put(url, data=params)
391
392
393			class FederatedCloudShare(WithRequester):
394			API_URL = "/ocs/v2.php/apps/files_sharing/api/v1"
395			FEDERATED = "remote_shares"
396
397			def get_federated_url(self, additional_url=""):
398			if additional_url:
399			return "/".join([self.FEDERATED, additional_url])
400			return self.FEDERATED
401
402			@nextcloud_method
403			def list_accepted_federated_cloudshares(self):
404			url = self.get_federated_url()
405			return self.requester.get(url)
406
407			@nextcloud_method
408			def get_known_federated_cloudshare(self, sid):
409			url = self.get_federated_url(sid)
410			return self.requester.get(url)
411
412			@nextcloud_method
413			def delete_accepted_federated_cloudshare(self, sid):
414			url = self.get_federated_url(sid)
415			return self.requester.delete(url)
416
417			@nextcloud_method
418			def list_pending_federated_cloudshares(self, sid):
419			url = self.get_federated_url("pending")
420			return self.requester.get(url)
421
422			@nextcloud_method
423			def accept_pending_federated_cloudshare(self, sid):
424			url = self.get_federated_url("pending/{sid}".format(sid=sid))
425			return self.requester.post(url)
426
427			@nextcloud_method
428			def decline_pending_federated_cloudshare(self, sid):
429			url = self.get_federated_url("pending/{sid}".format(sid=sid))
430			return self.requester.delete(url)
431
432
433			class Apps(WithRequester):
434			API_URL = "/ocs/v1.php/cloud/apps"
435
436			@nextcloud_method
437			def get_apps(self, filter=None):
438			"""
439			Get a list of apps installed on the Nextcloud server
440
441			:param filter: str, optional "enabled" or "disabled"
442			:return:
443			"""
444			params = {
445			"filter": filter
446			}
447			return self.requester.get(params=params)
448
449			@nextcloud_method
450			def get_app(self, app_id):
451			"""
452			Provide information on a specific application
453
454			:param app_id: str, app id
455			:return:
456			"""
457			return self.requester.get(app_id)
458
459			@nextcloud_method
460			def enable_app(self, app_id):
461			"""
462			Enable an app
463
464			:param app_id: str, app id
465			:return:
466			"""
467			return self.requester.post(app_id)
468
469			@nextcloud_method
470			def disable_app(self, app_id):
471			"""
472			Disable the specified app
473
474			:param app_id: str, app id
475			:return:
476			"""
477			return self.requester.delete(app_id)
478
479
480			class Group(WithRequester):
481			API_URL = "/ocs/v1.php/cloud/groups"
482
483			@nextcloud_method
484			def get_groups(self, search=None, limit=None, offset=None):
485			"""
486			Retrieve a list of groups from the Nextcloud server
487
488			:param search: string, optional search string
489			:param limit: int, optional limit value
490			:param offset: int, optional offset value
491			:return:
492			"""
493			params = {
494			'search': search,
495			'limit': limit,
496			'offset': offset
497			}
498			return self.requester.get(params=params)
499
500			@nextcloud_method
501			def add_group(self, gid):
502			"""
503			Add a new group
504
505			:param gid: str, group name
506			:return:
507			"""
508			msg = {"groupid": gid}
509			return self.requester.post("", msg)
510
511			@nextcloud_method
512			def get_group(self, gid):
513			"""
514			Retrieve a list of group members
515
516			:param gid: str, group name
517			:return:
518			"""
519			return self.requester.get("{gid}".format(gid=gid))
520
521			@nextcloud_method
522			def get_subadmins(self, gid):
523			"""
524			List subadmins of the group
525
526			:param gid: str, group name
527			:return:
528			"""
529			return self.requester.get("{gid}/subadmins".format(gid=gid))
530
531			@nextcloud_method
532			def delete_group(self, gid):
533			"""
534			Remove a group
535
536			:param gid: str, group name
537			:return:
538			"""
539			return self.requester.delete("{gid}".format(gid=gid))
540
541
542			class User(WithRequester):
543			API_URL = "/ocs/v1.php/cloud/users"
544
545			@nextcloud_method
546			def add_user(self, uid, passwd):
547			"""
548			Create a new user on the Nextcloud server
549
550			:param uid: str, uid of new user
551			:param passwd: str, password of new user
552			:return:
553			"""
554			msg = {'userid': uid, 'password': passwd}
555			return self.requester.post("", msg)
556
557			@nextcloud_method
558			def get_users(self, search=None, limit=None, offset=None):
559			"""
560			Retrieve a list of users from the Nextcloud server
561
562			:param search: string, optional search string
563			:param limit: int, optional limit value
564			:param offset: int, optional offset value
565			:return:
566			"""
567			params = {
568			'search': search,
569			'limit': limit,
570			'offset': offset
571			}
572			return self.requester.get(params=params)
573
574			@nextcloud_method
575			def get_user(self, uid):
576			"""
577			Retrieve information about a single user
578
579			:param uid: str, uid of user
580			:return:
581			"""
582			return self.requester.get("{uid}".format(uid=uid))
583
584			@nextcloud_method
585			def edit_user(self, uid, what, value):
586			"""
587			Edit attributes related to a user
588
589			Users are able to edit email, displayname and password; admins can also edit the quota value
590
591			:param uid: str, uid of user
592			:param what: str, the field to edit
593			:param value: str, the new value for the field
594			:return:
595			"""
596			what_to_key_map = dict(
597			email="email", quota="quote", phone="phone", address="address", website="website",
598			twitter="twitter", displayname="displayname", password="password",
599			)
600			assert what in what_to_key_map, (
601			"You have chosen to edit user's '{what}', but you can choose only from: {choices}"
602			.format(what=what, choices=", ".join(what_to_key_map.keys()))
603			)
604
605			url = "{uid}".format(uid=uid)
606			msg = dict(
607			key=what_to_key_map[what],
608			value=value,
609			)
610			return self.requester.put(url, msg)
611
612			@nextcloud_method
613			def disable_user(self, uid):
614			"""
615			Disable a user on the Nextcloud server so that the user cannot login anymore
616
617			:param uid: str, uid of user
618			:return:
619			"""
620			return self.requester.put("{uid}/disable".format(uid=uid))
621
622			@nextcloud_method
623			def enable_user(self, uid):
624			"""
625			Enable a user on the Nextcloud server so that the user can login again
626
627			:param uid: str, uid of user
628			:return:
629			"""
630			return self.requester.put("{uid}/enable".format(uid=uid))
631
632			@nextcloud_method
633			def delete_user(self, uid):
634			"""
635			Delete a user from the Nextcloud server
636
637			:param uid: str, uid of user
638			:return:
639			"""
640			return self.requester.delete("{uid}".format(uid=uid))
641
642			@nextcloud_method
643			def add_to_group(self, uid, gid):
644			"""
645			Add the specified user to the specified group
646
647			:param uid: str, uid of user
648			:param gid: str, name of group
649			:return:
650			"""
651			url = "{uid}/groups".format(uid=uid)
652			msg = {'groupid': gid}
653			return self.requester.post(url, msg)
654
655			@nextcloud_method
656			def remove_from_group(self, uid, gid):
657			"""
658			Remove the specified user from the specified group
659
660			:param uid: str, uid of user
661			:param gid: str, name of group
662			:return:
663			"""
664			url = "{uid}/groups".format(uid=uid)
665			msg = {'groupid': gid}
666			return self.requester.delete(url, msg)
667
668			@nextcloud_method
669			def create_subadmin(self, uid, gid):
670			"""
671			Make a user the subadmin of a group
672
673			:param uid: str, uid of user
674			:param gid: str, name of group
675			:return:
676			"""
677			url = "{uid}/subadmins".format(uid=uid)
678			msg = {'groupid': gid}
679			return self.requester.post(url, msg)
680
681			@nextcloud_method
682			def remove_subadmin(self, uid, gid):
683			"""
684			Remove the subadmin rights for the user specified from the group specified
685
686			:param uid: str, uid of user
687			:param gid: str, name of group
688			:return:
689			"""
690			url = "{uid}/subadmins".format(uid=uid)
691			msg = {'groupid': gid}
692			return self.requester.delete(url, msg)
693
694			@nextcloud_method
695			def get_subadmin_groups(self, uid):
696			"""
697			Get the groups in which the user is a subadmin
698
699			:param uid: str, uid of user
700			:return:
701			"""
702			url = "{uid}/subadmins".format(uid=uid)
703			return self.requester.get(url)
704
705			@nextcloud_method
706			def resend_welcome_mail(self, uid):
707			"""
708			Trigger the welcome email for this user again
709
710			:param uid: str, uid of user
711			:return:
712			"""
713			url = "{uid}/welcome".format(uid=uid)
714			return self.requester.post(url)
715
716
717			class Activity(WithRequester):
718			API_URL = "/ocs/v2.php/apps/activity/api/v2/activity"
719
720			@nextcloud_method
721			def get_activities(self, since=None, limit=None, object_type=None, object_id=None, sort=None):
722			"""
723			Get an activity feed showing your file changes and other interesting things going on in your Nextcloud
724
725			All args are optional
726
727			Args:
728			since (int): ID of the last activity that you’ve seen
729			limit (int): How many activities should be returned (default: 50)
730			object_type (string): Filter the activities to a given object. May only appear together with object_id
731			object_id (string): Filter the activities to a given object. May only appear together with object_type
732			sort (str, "asc" or "desc"): Sort activities ascending or descending (from the since) (Default: desc)
733
734			Returns:
735
736			"""
737			params = dict(
738			since=since,
739			limit=limit,
740			object_type=object_type,
741			object_id=object_id,
742			sort=sort
743			)
744			if params['object_type'] and params['object_id']:
745			return self.requester.get(url="filter", params=params)
746			return self.requester.get(params=params)
747
748
749			class OCSCode(enum.IntEnum):
750			OK = 100
751			SERVER_ERROR = 996
752			NOT_AUTHORIZED = 997
753			NOT_FOUND = 998
754			UNKNOWN_ERROR = 999
755
756
757			class ShareType(enum.IntEnum):
758			USER = 0
759			GROUP = 1
760			PUBLIC_LINK = 3
761			FEDERATED_CLOUD_SHARE = 6
762
763
764			class Permission(enum.IntEnum):
765			""" Permission for Share have to be sum of selected permissions """
766			READ = 1
767			UPDATE = 2
768			CREATE = 4
769			DELETE = 8
770			SHARE = 16
771			ALL = 31
772
773
774			QUOTA_UNLIMITED = -3
775
776
777			def datetime_to_expire_date(date):
778			return date.strftime("%Y-%m-%d")
779

EnterpriseyIntranet / nextcloud-API

Push — master ( 52addc...0b5289 )

NextCloud.Group.get_subadmins() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like