User.streams() - Code Metrics - Inspection of "Added user.streams() function to list streams" - connectordb/connectordb-python - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d5d9b4...ed39b3 )

by Daniel

created 2017-01-03 06:21 UTC

User.streams() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
c	0
b	0
f	0
dl	0
loc	20
rs	9.2

from __future__ import absolute_import
import json
import os

from ._connectorobject import ConnectorObject


class User(ConnectorObject):

    def create(self, email, password, role="user", public=True, **kwargs):
        """Creates the given user - using the passed in email and password.

        You can also set other default properties by passing in the relevant information::

            usr.create("my@email","mypass",description="I like trains.")

        Furthermore, ConnectorDB permits immediate initialization of an entire user tree,
        so that you can create all relevant devices and streams in one go::

            usr.create("my@email","mypass",devices={
                "device1": {
                    "nickname": "My train",
                    "streams": {
                        "stream1": {
                            "schema": "{\"type\":\"string\"}",
                            "datatype": "train.choochoo"
                        }
                    },
                }
            })

        The user and meta devices are created by default. If you want to add streams to the user device,
        use the "streams" option in place of devices in create.
        """
        kwargs["email"] = email
        kwargs["password"] = password
        kwargs["role"] = role
        kwargs["public"] = public
        self.metadata = self.db.create(
            self.path, kwargs).json()

    def set_password(self, new_password):
        """Sets a new password for the user"""

        self.set({"password": new_password})

    def devices(self):
        """Returns the list of devices that belong to the user"""
        result = self.db.read(self.path, {"q": "ls"})

        if result is None or result.json() is None:
            return []
        devices = []
        for d in result.json():
            dev = self[d["name"]]
            dev.metadata = d
            devices.append(dev)
        return devices

    def streams(self, public=False, downlink=False, visible=True):
        """Returns the list of streams that belong to the user.
        The list can optionally be filtered in 3 ways:
            - public: when True, returns only streams belonging to public devices
            - downlink: If True, returns only downlink streams
            - visible: If True (default), returns only streams of visible devices
        """
        result = self.db.read(self.path, {"q": "streams",
                                          "public": str(public).lower(),
                                          "downlink": str(downlink).lower(),
                                          "visible": str(visible).lower()})

        if result is None or result.json() is None:
            return []
        streams = []
        for d in result.json():
            s = self[d["device"]][d["name"]]
            s.metadata = d
            streams.append(s)
        return streams

    def __getitem__(self, device_name):
        """Gets the child device by name"""
        return Device(self.db, self.path + "/" + device_name)

    def __repr__(self):
        """Returns a string representation of the user"""
        return "[User:%s]" % (self.path, )

    def export(self, directory):
        """Exports the ConnectorDB user into the given directory.
        The resulting export can be imported by using the import command(cdb.import(directory)),

        Note that Python cannot export passwords, since the REST API does
        not expose password hashes. Therefore, the imported user will have
        password same as username.

        The user export function is different than device and stream exports because
        it outputs a format compatible directly with connectorDB's import functionality:

            connectordb import < mydatabase > <directory >

        This also means that you can export multiple users into the same directory without issue
        """

        exportInfoFile = os.path.join(directory, "connectordb.json")
        if os.path.exists(directory):
            # Ensure that there is an export there already, and it is version 1
            if not os.path.exists(exportInfoFile):
                raise FileExistsError(
                    "The export directory already exsits, and is not a ConnectorDB export.")
            with open(exportInfoFile) as f:
                exportInfo = json.load(f)
            if exportInfo["Version"] != 1:
                raise ValueError(
                    "Could not export to directory: incompatible export versions.")
        else:
            # The folder doesn't exist. Make it.
            os.mkdir(directory)

            with open(exportInfoFile, "w") as f:
                json.dump(
                    {"Version": 1, "ConnectorDB": self.db.get("meta/version").text}, f)

        # Now we create the user directory
        udir = os.path.join(directory, self.name)
        os.mkdir(udir)

        # Write the user's info
        with open(os.path.join(udir, "user.json"), "w") as f:
            json.dump(self.data, f)

        # Now export the devices one by one
        for d in self.devices():
            d.export(os.path.join(udir, d.name))

    def import_device(self, directory):
        """Imports a device from the given directory. You export the device
        by using device.export()

        There are two special cases: user and meta devices.
        If the device name is meta, import_device will not do anything.
        If the device name is "user", import_device will overwrite the user device
        even if it exists already.
        """

        # read the device's info
        with open(os.path.join(directory, "device.json"), "r") as f:
            ddata = json.load(f)

        d = self[ddata["name"]]

        dname = ddata["name"]
        del ddata["name"]

        if dname == "meta":
            return
        elif dname == "user":
            d.set(ddata)
        elif d.exists():
            raise ValueError("The device " + d.name + " already exists")
        else:
            d.create(**ddata)

        # Now import all of the streams
        for name in os.listdir(directory):
            sdir = os.path.join(directory, name)
            if os.path.isdir(sdir):
                d.import_stream(sdir)

    # -----------------------------------------------------------------------
    # Following are getters and setters of the user's properties

    @property
    def email(self):
        """gets the user's email address"""
        if "email" in self.data:
            return self.data["email"]
        return None

    @email.setter
    def email(self, new_email):
        """sets the user's email address"""
        self.set({"email": new_email})

    @property
    def public(self):
        """gets whether the user is public
        (this means different things based on connectordb permissions setup - connectordb.com
        has this be whether the user is publically visible. Devices are individually public / private.)
        """
        if "public" in self.data:
            return self.data["public"]
        return None

    @public.setter
    def public(self, new_public):
        """Attempts to set whether the user is public"""
        self.set({"public": new_public})

    @property
    def role(self):
        """Gets the role of the user. This is the permissions level that the user has. It might
        not be accessible depending on the permissions setup of ConnectorDB. Returns None if not accessible"""
        if "role" in self.data:
            return self.data["role"]
        return None

    @role.setter
    def role(self, new_role):
        """ Attempts to set the user's role"""
        self.set({"role": new_role})

# The import has to go on the bottom because py3 imports are annoying
from ._device import Device


1		from __future__ import absolute_import
2		import json
3		import os
4
5		from ._connectorobject import ConnectorObject
6
7
8		class User(ConnectorObject):
9
10		def create(self, email, password, role="user", public=True, **kwargs):
11		"""Creates the given user - using the passed in email and password.
12
13		You can also set other default properties by passing in the relevant information::
14
15		usr.create("my@email","mypass",description="I like trains.")
16
17		Furthermore, ConnectorDB permits immediate initialization of an entire user tree,
18		so that you can create all relevant devices and streams in one go::
19
20		usr.create("my@email","mypass",devices={
21		"device1": {
22		"nickname": "My train",
23		"streams": {
24		"stream1": {
25		"schema": "{\"type\":\"string\"}",
26		"datatype": "train.choochoo"
27		}
28		},
29		}
30		})
31
32		The user and meta devices are created by default. If you want to add streams to the user device,
33		use the "streams" option in place of devices in create.
34		"""
35		kwargs["email"] = email
36		kwargs["password"] = password
37		kwargs["role"] = role
38		kwargs["public"] = public
39		self.metadata = self.db.create(
40		self.path, kwargs).json()
41
42		def set_password(self, new_password):
43	View Code Duplication	"""Sets a new password for the user"""
		0 ignored issues – show Duplication introduced 2016-05-26 11:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
44		self.set({"password": new_password})
45
46		def devices(self):
47		"""Returns the list of devices that belong to the user"""
48		result = self.db.read(self.path, {"q": "ls"})
49
50		if result is None or result.json() is None:
51		return []
52		devices = []
53		for d in result.json():
54		dev = self[d["name"]]
55		dev.metadata = d
56		devices.append(dev)
57		return devices
58
59		def streams(self, public=False, downlink=False, visible=True):
60		"""Returns the list of streams that belong to the user.
61		The list can optionally be filtered in 3 ways:
62		- public: when True, returns only streams belonging to public devices
63		- downlink: If True, returns only downlink streams
64		- visible: If True (default), returns only streams of visible devices
65		"""
66		result = self.db.read(self.path, {"q": "streams",
67		"public": str(public).lower(),
68		"downlink": str(downlink).lower(),
69		"visible": str(visible).lower()})
70
71		if result is None or result.json() is None:
72		return []
73		streams = []
74		for d in result.json():
75		s = self[d["device"]][d["name"]]
76		s.metadata = d
77		streams.append(s)
78		return streams
79
80		def __getitem__(self, device_name):
81		"""Gets the child device by name"""
82		return Device(self.db, self.path + "/" + device_name)
83
84		def __repr__(self):
85		"""Returns a string representation of the user"""
86		return "[User:%s]" % (self.path, )
87
88		def export(self, directory):
89		"""Exports the ConnectorDB user into the given directory.
90		The resulting export can be imported by using the import command(cdb.import(directory)),
91
92		Note that Python cannot export passwords, since the REST API does
93		not expose password hashes. Therefore, the imported user will have
94		password same as username.
95
96		The user export function is different than device and stream exports because
97		it outputs a format compatible directly with connectorDB's import functionality:
98
99		connectordb import < mydatabase > <directory >
100
101		This also means that you can export multiple users into the same directory without issue
102		"""
103
104		exportInfoFile = os.path.join(directory, "connectordb.json")
105		if os.path.exists(directory):
106		# Ensure that there is an export there already, and it is version 1
107		if not os.path.exists(exportInfoFile):
108		raise FileExistsError(
109		"The export directory already exsits, and is not a ConnectorDB export.")
110		with open(exportInfoFile) as f:
111		exportInfo = json.load(f)
112		if exportInfo["Version"] != 1:
113		raise ValueError(
114		"Could not export to directory: incompatible export versions.")
115		else:
116		# The folder doesn't exist. Make it.
117		os.mkdir(directory)
118
119		with open(exportInfoFile, "w") as f:
120		json.dump(
121		{"Version": 1, "ConnectorDB": self.db.get("meta/version").text}, f)
122
123		# Now we create the user directory
124		udir = os.path.join(directory, self.name)
125		os.mkdir(udir)
126
127		# Write the user's info
128		with open(os.path.join(udir, "user.json"), "w") as f:
129		json.dump(self.data, f)
130
131		# Now export the devices one by one
132		for d in self.devices():
133		d.export(os.path.join(udir, d.name))
134
135		def import_device(self, directory):
136		"""Imports a device from the given directory. You export the device
137		by using device.export()
138
139		There are two special cases: user and meta devices.
140		If the device name is meta, import_device will not do anything.
141		If the device name is "user", import_device will overwrite the user device
142		even if it exists already.
143		"""
144
145		# read the device's info
146		with open(os.path.join(directory, "device.json"), "r") as f:
147		ddata = json.load(f)
148
149		d = self[ddata["name"]]
150
151		dname = ddata["name"]
152		del ddata["name"]
153
154		if dname == "meta":
155		return
156		elif dname == "user":
157		d.set(ddata)
158		elif d.exists():
159		raise ValueError("The device " + d.name + " already exists")
160		else:
161		d.create(**ddata)
162
163		# Now import all of the streams
164		for name in os.listdir(directory):
165		sdir = os.path.join(directory, name)
166		if os.path.isdir(sdir):
167		d.import_stream(sdir)
168
169		# -----------------------------------------------------------------------
170		# Following are getters and setters of the user's properties
171
172		@property
173		def email(self):
174		"""gets the user's email address"""
175		if "email" in self.data:
176		return self.data["email"]
177		return None
178
179		@email.setter
180		def email(self, new_email):
181		"""sets the user's email address"""
182		self.set({"email": new_email})
183
184		@property
185		def public(self):
186		"""gets whether the user is public
187		(this means different things based on connectordb permissions setup - connectordb.com
188		has this be whether the user is publically visible. Devices are individually public / private.)
189		"""
190		if "public" in self.data:
191		return self.data["public"]
192		return None
193
194		@public.setter
195		def public(self, new_public):
196		"""Attempts to set whether the user is public"""
197		self.set({"public": new_public})
198
199		@property
200		def role(self):
201		"""Gets the role of the user. This is the permissions level that the user has. It might
202		not be accessible depending on the permissions setup of ConnectorDB. Returns None if not accessible"""
203		if "role" in self.data:
204		return self.data["role"]
205		return None
206
207		@role.setter
208		def role(self, new_role):
209		""" Attempts to set the user's role"""
210		self.set({"role": new_role})
211
212		# The import has to go on the bottom because py3 imports are annoying
213		from ._device import Device
214

connectordb / connectordb-python

GitHub Access Token became invalid

Push — master ( d5d9b4...ed39b3 )

User.streams() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like