NextCloud.Share.get_shares_from_path() - Code Metrics - Inspection of "WebDAV" - EnterpriseyIntranet/nextcloud-API - Measure and Improve Code Quality continuously with Scrutinizer

Passed
Pull Request — master (#18)

unknown
created 2019-01-14 00:50 UTC
NextCloud.Share.get_shares_from_path() A

↳ Parent: NextCloud
Complexity

Conditions
Size

Total Lines	20
Code Lines	8
Duplication

Lines	0
Ratio	0 %
Importance

Changes
Metric	Value
eloc	8
dl	0
loc	20
rs	10
c	0
b	0
f	0
cc	3
nop	4
import re
import enum
import requests
import xml.etree.ElementTree as ET
import os

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


class WebDAVRequester(Requester):

    def __init__(self, *args, **kwargs):
        super(WebDAVRequester, self).__init__(*args, **kwargs)

    def propfind(self, additional_url="", headers=None):
        url = self.get_full_url(additional_url=additional_url)
        res = requests.request('PROPFIND', url, auth=self.auth_pk, headers=headers)
        return self.rtn(res)

    def download(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 res

    def make_collection(self, additional_url=""):
        url = self.get_full_url(additional_url=additional_url)
        res = requests.request("MKCOL", url=url, auth=self.auth_pk)
        return res

    def move(self, url, destination, overwrite=False):
        url = self.get_full_url(additional_url=url)
        destionation_url = self.get_full_url(additional_url=destination)
        headers = {
            "Destination": destionation_url,
            "Overwrite": "T" if overwrite else "F"
        }
        res = requests.request("MOVE", url=url, auth=self.auth_pk, headers=headers)
        return res

    def copy(self, url, destination, overwrite=False):
        url = self.get_full_url(additional_url=url)
        destionation_url = self.get_full_url(additional_url=destination)
        headers = {
            "Destination": destionation_url,
            "Overwrite": "T" if overwrite else "F"
        }
        res = requests.request("COPY", url=url, auth=self.auth_pk, headers=headers)
        return res


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)
        webdav_requester = WebDAVRequester(endpoint, user, password)

        self.functionality = {
            "Apps": Apps(requester),
            "Group": Group(requester),
            "GroupFolders": GroupFolders(requester),
            "Share": Share(requester),
            "User": User(requester),
            "FederatedCloudShare": FederatedCloudShare(requester),
            "WebDAV": WebDAV(webdav_requester, js=js),
        }
        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 WebDAV(WithRequester):

    API_URL = "/remote.php/dav/files"

    def __init__(self, *args, **kwargs):
        super(WebDAV, self).__init__(*args)
        self.js = kwargs.get('js')

    @nextcloud_method
    def list_folders(self, uid, path=None, depth=1):
        """
        Get path files list with files properties for given user, with given depth

        Args:
            uid (str): uid of user
            path (str/None): files path
            depth: depth of listing files (directories content for example)

        Returns:
            list of dicts if js
            list of File objects if not js
        """
        # TODO: add possibility to request additional properties (in a request body)
        #  https://docs.nextcloud.com/server/14/developer_manual/client_apis/WebDAV/basic.html#requesting-properties
        additional_url = uid
        if path:
            additional_url = "{}/{}".format(additional_url, path)
        resp = self.requester.propfind(additional_url=additional_url, headers={"Depth": str(depth)})
        response_xml_data = ET.fromstring(resp)
        files_data = [File(single_file) for single_file in response_xml_data]
        return files_data if not self.js else [each.as_dict() for each in files_data]

    @nextcloud_method
    def download_file(self, uid, path):
        """
        Download file of given user by path

        File will be saved to working directory
        path argument must be valid file path

        Exception will be raised if:
            path doesn't exist
            path is a directory
            file with same name already exists in working directory

        Args:
            uid (str): uid of user
            path (str): file path


        Returns:
            None

        """
        additional_url = "/".join([uid, path])
        filename = path.split('/')[-1] if '/' in path else path
        file_data = self.list_folders(uid=uid, path=path, depth=0)[0]
        if file_data.status_code == 404:
            raise ValueError("Given path doesn't exist")
        file_resource_type = file_data.get('resource_type') if self.js else file_data.resource_type
        if file_resource_type == File.COLLECTION_RESOURCE_TYPE:
            raise ValueError("This is a collection, please specify file path")
        if filename in os.listdir('./'):
            raise ValueError("File with such name already exists in this directory")
        res = self.requester.download(additional_url)
        with open(filename, 'wb') as f:
            f.write(res.content)

    @nextcloud_method
    def upload_file(self, uid, local_filepath, remote_filepath):
        """
        Upload file to Nextcloud storage

        Args:
            uid (str): uid of user
            local_filepath (str): path to file on local storage
            remote_filepath (str): path where to upload file on Nextcloud storage

        Returns:

        """
        with open(local_filepath, 'rb') as f:
            file_content = f.read()
        additional_url = "/".join([uid, remote_filepath])
        return self.requester.put(additional_url, data=file_content)

    @nextcloud_method
    def create_folder(self, uid, folder_path):
        """
        Create folder on Nextcloud storage

        Args:
            uid (str): uid of user
            folder_path (str): folder path

        Returns:

        """
        return self.requester.make_collection(additional_url="/".join([uid, folder_path]))

    @nextcloud_method
    def delete_path(self, uid, path):
        """
        Delete file or folder with all content of given user by path

        Args:
            uid (str): uid of user
            path (str): file or folder path to delete

        Returns:

        """
        url = "/".join([uid, path])
        return self.requester.delete(url=url)

    @nextcloud_method
    def move_path(self, uid, path, destination_path, overwrite=False):
        """
        Move file or folder to destination

        Args:
            uid (str): uid of user
            path (str): file or folder path to move
            destionation_path (str): destination where to move
            overwrite (bool): allow destination path overriding

        Returns:

        """
        path_url = "/".join([uid, path])
        destination_path_url = "/".join([uid, destination_path])
        return self.requester.move(url=path_url, destination=destination_path_url, overwrite=overwrite)

    @nextcloud_method
    def copy_path(self, uid, path, destination_path, overwrite=False):
        """
        Copy file or folder to destination

        Args:
            uid (str): uid of user
            path (str): file or folder path to copy
            destionation_path (str): destination where to copy
            overwrite (bool): allow destination path overriding

        Returns:

        """
        path_url = "/".join([uid, path])
        destination_path_url = "/".join([uid, destination_path])
        return self.requester.copy(url=path_url, destination=destination_path_url, overwrite=overwrite)

    @nextcloud_method
    def set_favorites(self):
        pass

    @nextcloud_method
    def list_favorites(self):
        pass


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


class File(object):

    SUCCESS_STATUS = 'HTTP/1.1 200 OK'

    FILE_PROPERTIES = {
        "getlastmodified": "last_modified",
        "getetag": "etag",
        "getcontenttype": "content_type",
        "resourcetype": "resource_type",
        "getcontentlength": "content_length",
        "id": "id",
        "fileid": "file_id",
        "favorite": "favorite",
        "comments-href": "comments_href",
        "comments-count": "comments_count",
        "comments-unread": "comments_unread",
        "owner-id": "owner_id",
        "owner-display-name": "owner_display_name",
        "share-types": "share_types",
        "checksums": "check_sums",
        "has-preview": "has_preview",
        "size": "size",
        "href": "href"
    }
    COLLECTION_RESOURCE_TYPE = 'collection'

    def __init__(self, xml_data):
        self.href = xml_data.find('{DAV:}href').text
        for propstat in xml_data.iter('{DAV:}propstat'):
            if propstat.find('{DAV:}status').text != self.SUCCESS_STATUS:
                continue
            for file_property in propstat.find('{DAV:}prop'):
                file_property_name = re.sub("{.*}", "", file_property.tag)
                if file_property_name not in self.FILE_PROPERTIES:
                    continue
                if file_property_name == 'resourcetype':
                    value = self._extract_resource_type(file_property)
                else:
                    value = file_property.text
                setattr(self, self.FILE_PROPERTIES[file_property_name], value)

    def _extract_resource_type(self, file_property):
        file_type = list(file_property)
        if file_type:
            return re.sub("{.*}", "", file_type[0].tag)
        return None

    def as_dict(self):
        return {key: value for key, value in self.__dict__.items() if key in self.FILE_PROPERTIES.values()}


QUOTA_UNLIMITED = -3


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

EnterpriseyIntranet / nextcloud-API

Pull Request — master (#18)

NextCloud.Share.get_shares_from_path() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like
