NextCloud.GroupFolders.set_access_to_group_folder() - Code Metrics - Inspection of "Merge pull request #11 from danil-topc/group_folde..." - EnterpriseyIntranet/nextcloud-API - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( b3f222...0c1f9d )

by Matěj

created 2019-01-07 22:35 UTC

older() A

↳ Parent: NextCloud.GroupFolders.create_group_folder()

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	4
rs	10
c	0
b	0
f	0
cc	1
nop	4

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, passwd, js=False):
        self.query_components = []

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

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

EnterpriseyIntranet / nextcloud-API

Push — master ( b3f222...0c1f9d )

older() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like