User.get_number_new_users() - Code Metrics - Inspection of "display number of new users in admin page" - um-cseg/chez-betty - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 1c4fdd...cd839c )

unknown

created 2016-09-04 19:00 UTC

User.get_number_new_users() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
dl	0
loc	11
rs	9.4285
c	0
b	0
f	0

from pyramid.security import authenticated_userid
import hashlib
import binascii
import os
from .model import *
from . import account
from . import event
from . import request
from . import request_post
from chezbetty import utility

import ldap3
import random
import string

class InvalidUserException(Exception):
    pass


class LDAPLookup(object):
    """class that allows lookup of an individual in the UM directory
    based on Michigan ID number, uniqname, MCARD"""

    SERVER = None
    USERNAME = None
    BASE_DN = "ou=People,dc=umich,dc=edu"
    PASSWORD = None
    ATTRIBUTES = ["uid", "entityid", "displayName"]
    # http://www.itcs.umich.edu/itcsdocs/r1463/attributes-for-ldap.html
    DETAILS_ATTRIBUTES = [
            # Could be interesting but require extra perm's from ITS
            #"umichInstRoles",
            #"umichAlumStatus",
            #"umichAAAcadProgram",
            #"umichAATermStatus",
            #"umichHR",
            "notice",
            "ou",
            "umichDescription",
            "umichTitle",
            ]

    def __init__(self):
        self.__conn = None

    def __connect(self):
        if not self.__conn:
            s = ldap3.Server(self.SERVER, port=636, use_ssl=True, get_info=ldap3.GET_ALL_INFO)
            self.__conn = ldap3.Connection(s, auto_bind=True,
                    user=self.USERNAME, password=self.PASSWORD,
                    client_strategy=ldap3.STRATEGY_SYNC,
                    authentication=ldap3.AUTH_SIMPLE
            )


    def __do_lookup(self, k, v, attributes, full_dict=False):
        self.__connect()
        query = "(%s=%s)" % (k, v)
        try:
            self.__conn.search(self.BASE_DN,
                    query,
                    ldap3.SEARCH_SCOPE_WHOLE_SUBTREE,
                    attributes=attributes
            )
        except:
            # sometimes our connections time out
            self.__conn = None
            self.__connect()
            self.__conn.search(self.BASE_DN,
                    query,
                    ldap3.SEARCH_SCOPE_WHOLE_SUBTREE,
                    attributes=attributes
            )

        if len(self.__conn.response) == 0:
            raise InvalidUserException()

        if full_dict:
            return self.__conn.response[0]["attributes"]

        return {
            "umid":self.__conn.response[0]["attributes"]["entityid"],
            "uniqname":self.__conn.response[0]["attributes"]["uid"][0],
            "name":self.__conn.response[0]["attributes"]["displayName"][0]
        }

    def __lookup(self, k, v):
        return self.__do_lookup(k, v, self.ATTRIBUTES)

    def __detail_lookup(self, k, v):
        return self.__do_lookup(k, v,
                self.ATTRIBUTES + self.DETAILS_ATTRIBUTES,
                full_dict=True,
                )

    def lookup_umid(self, umid, details=False):
        if details:
            return self.__detail_lookup("entityid", umid)
        else:
            return self.__lookup("entityid", umid)

    def lookup_uniqname(self, uniqname, details=False):
        if details:
            return self.__detail_lookup("uid", uniqname)
        else:
            return self.__lookup("uid", uniqname)


class User(account.Account):
    __tablename__ = 'users'
    __mapper_args__ = {'polymorphic_identity': 'user'}

    id        = Column(Integer, ForeignKey("accounts.id"), primary_key=True)
    uniqname  = Column(String(8), nullable=False, unique=True)
    umid      = Column(String(8), unique=True)
    _password = Column("password", String(255))
    _salt     = Column("salt", String(255))
    enabled   = Column(Boolean, nullable=False, default=True)
    archived  = Column(Boolean, nullable=False, default=False)
    role      = Column(Enum("user", "serviceaccount", "manager", "administrator", name="user_type"),
                       nullable=False, default="user")

    administrative_events = relationship(event.Event, foreign_keys=[event.Event.user_id], backref="admin")
    events_deleted        = relationship(event.Event, foreign_keys=[event.Event.deleted_user_id], backref="deleted_user")
    requests              = relationship(request.Request, foreign_keys=[request.Request.user_id], backref="user")
    request_posts         = relationship(
                              request_post.RequestPost,
                              foreign_keys=[request_post.RequestPost.user_id],
                              backref="user",
                            )

    __ldap = LDAPLookup()

    def __init__(self, uniqname, umid, name):
        self.enabled = True
        self.uniqname = uniqname
        self.umid = umid
        self.name = name
        self.balance = 0.0

    def __str__(self):
        return "<User: id {}, uniqname {}, umid {}, name {}, balance {}>".\
                format(self.id, self.uniqname, self.umid, self.name, self.balance)

    @classmethod
    def from_id(cls, id):
        return DBSession.query(cls).filter(cls.id == id).one()

    def get_details(self):
        return self.__ldap.lookup_uniqname(self.uniqname, details=True)

    @classmethod
    def from_uniqname(cls, uniqname, local_only=False):
        u = DBSession.query(cls).filter(cls.uniqname == uniqname).first()
        if not u and not local_only:
            u = cls(**cls.__ldap.lookup_uniqname(uniqname))
            DBSession.add(u)
        return u

    @classmethod
    def from_umid(cls, umid, create_if_never_seen=False):
        u = DBSession.query(cls).filter(cls.umid == umid).first()
        if not u:
            if create_if_never_seen:
                u = cls(**cls.__ldap.lookup_umid(umid))
                DBSession.add(u)
                utility.new_user_email(u)
            else:
                raise InvalidUserException()
        return u

    @classmethod
    def from_fuzzy(cls, search_str, any=True):
        q = DBSession.query(cls)\
                     .filter(or_(
                            cls.uniqname.ilike('%{}%'.format(search_str)),
                            cls.umid.ilike('%{}%'.format(search_str)),
                            cls.name.ilike('%{}%'.format(search_str))
                      ))
        if not any:
            q = q.filter(cls.enabled)\
                 .filter(cls.archived==False)

        return q.all()

    @classmethod
    def all(cls):
        return DBSession.query(cls)\
                        .filter(cls.enabled)\
                        .order_by(cls.name)\
                        .all()

    @classmethod
    def count(cls):
        return DBSession.query(func.count(cls.id).label('c'))\
                        .filter(cls.role != 'serviceaccount')\
                        .filter(cls.archived == False)\
                        .filter(cls.enabled == True)\
                        .one().c

    @classmethod
    def get_admins(cls):
        return DBSession.query(cls)\
                        .filter(cls.enabled)\
                        .filter(cls.role=='administrator').all()

    @classmethod
    def get_shame_users(cls):
        return DBSession.query(cls)\
                        .filter(cls.enabled)\
                        .filter(cls.balance < -5)\
                        .order_by(cls.balance).all()

    @classmethod
    def get_normal_users(cls):
        return DBSession.query(cls)\
                        .filter(cls.enabled)\
                        .filter(cls.archived == False)\
                        .order_by(cls.name).all()

    @classmethod
    def get_archived_users(cls):
        return DBSession.query(cls)\
                        .filter(cls.enabled)\
                        .filter(cls.archived == True)\
                        .order_by(cls.name).all()

    @classmethod
    def get_disabled_users(cls):
        return DBSession.query(cls)\
                        .filter(cls.enabled == False)\
                        .order_by(cls.name).all()

    @classmethod
    def get_number_new_users(cls, start=None, end=None):
        r = DBSession.query(cls)\
                        .filter(cls.enabled == True)

        if start:
            r = r.filter(cls.created_at>=start)
        if end:
            r = r.filter(cls.created_at<end)

        return r.count()

    @classmethod
    def get_users_total(cls):
        return DBSession.query(func.sum(User.balance).label("total_balance"))\
                        .one().total_balance or Decimal(0.0)

    # Sum the total amount of money in user accounts that we are holding for
    # users. This is different from just getting the total because it doesn't
    # count users with negative balances
    @classmethod
    def get_amount_held(cls):
        return DBSession.query(func.sum(User.balance).label("total_balance"))\
                        .filter(User.balance>0)\
                        .one().total_balance or Decimal(0.0)

    @classmethod
    def get_amount_owed(cls):
        return DBSession.query(func.sum(User.balance).label("total_balance"))\
                        .filter(User.balance<0)\
                        .one().total_balance or Decimal(0.0)

    @classmethod
    def get_user_count_cumulative(cls):
        rows = DBSession.query(cls.created_at)\
                        .order_by(cls.created_at)\
                        .all()
        return utility.timeseries_cumulative(rows)

    def iterate_recent_items(self, limit=None, allow_duplicates=False):
        items = set()
        count = 0
        for e in self.events:
            if e.type == 'purchase':
                for transaction in e.transactions:
                    if transaction.type == 'purchase':
                        for line_item in transaction.subtransactions:
                            if (line_item.item not in items) or allow_duplicates:
                                count += 1
                                if limit is not None and count > limit:
                                    return
                                yield line_item.item
                                items.add(line_item.item)

    def __make_salt(self):
        return binascii.b2a_base64(open("/dev/urandom", "rb").read(32))[:-3].decode("ascii")

    @hybrid_property
    def password(self):
        return self._password

    @password.setter
    def password(self, password):
        if password == '':
            # Use this to clear the password so the user can't login
            self._password = None
        else:
            self._salt = self.__make_salt()
            salted = (self._salt + password).encode('utf-8')
            self._password = hashlib.sha256(salted).hexdigest()

    def random_password(self):
        password = ''.join(random.choice(string.ascii_letters + string.digits) for x in range(6))
        self._salt = self.__make_salt()
        salted = (self._salt + password).encode('utf-8')
        self._password = hashlib.sha256(salted).hexdigest()
        return password

    def check_password(self, cand):
        if not self._salt:
            return False
        salted = (self._salt + cand).encode('utf-8')
        c = hashlib.sha256(salted).hexdigest()
        return c == self._password

    @property
    def has_password(self):
        return self._password != None

    # Cash deposit limit is now fixed at $2 because we have a bill acceptor
    @property
    def deposit_limit(self):
        return 2.0


def get_user(request):
    login = authenticated_userid(request)
    if not login:
        return None
    return DBSession.query(User).filter(User.uniqname == login).one()


# This is in a stupid place due to circular input problems
@property
def __user_from_foreign_key(self):
    return User.from_id(self.user_id)
event.Event.user = __user_from_foreign_key


def groupfinder(userid, request):
    user = User.from_uniqname(userid)
    if user.role == "user":
        return ["user",]
    elif user.role == "manager":
        return ["user","manager"]
    elif user.role == "administrator":
        return ["user","manager","admin","serviceaccount"]
    elif user.role == "serviceaccount":
        return ["serviceaccount"]


1			from pyramid.security import authenticated_userid
2			import hashlib
3			import binascii
4			import os
5			from .model import *
6			from . import account
7			from . import event
8			from . import request
9			from . import request_post
10			from chezbetty import utility
11
12			import ldap3
13			import random
14			import string
15
16			class InvalidUserException(Exception):
17			pass
18
19
20			class LDAPLookup(object):
21			"""class that allows lookup of an individual in the UM directory
22			based on Michigan ID number, uniqname, MCARD"""
23
24			SERVER = None
25			USERNAME = None
26			BASE_DN = "ou=People,dc=umich,dc=edu"
27			PASSWORD = None
28			ATTRIBUTES = ["uid", "entityid", "displayName"]
29			# http://www.itcs.umich.edu/itcsdocs/r1463/attributes-for-ldap.html
30			DETAILS_ATTRIBUTES = [
31			# Could be interesting but require extra perm's from ITS
32			#"umichInstRoles",
33			#"umichAlumStatus",
34			#"umichAAAcadProgram",
35			#"umichAATermStatus",
36			#"umichHR",
37			"notice",
38			"ou",
39			"umichDescription",
40			"umichTitle",
41			]
42
43			def __init__(self):
44			self.__conn = None
45
46			def __connect(self):
47			if not self.__conn:
48			s = ldap3.Server(self.SERVER, port=636, use_ssl=True, get_info=ldap3.GET_ALL_INFO)
49			self.__conn = ldap3.Connection(s, auto_bind=True,
50			user=self.USERNAME, password=self.PASSWORD,
51			client_strategy=ldap3.STRATEGY_SYNC,
52			authentication=ldap3.AUTH_SIMPLE
53			)
54
55
56			def __do_lookup(self, k, v, attributes, full_dict=False):
57			self.__connect()
58			query = "(%s=%s)" % (k, v)
59			try:
60			self.__conn.search(self.BASE_DN,
61			query,
62			ldap3.SEARCH_SCOPE_WHOLE_SUBTREE,
63			attributes=attributes
64			)
65			except:
66			# sometimes our connections time out
67			self.__conn = None
68			self.__connect()
69			self.__conn.search(self.BASE_DN,
70			query,
71			ldap3.SEARCH_SCOPE_WHOLE_SUBTREE,
72			attributes=attributes
73			)
74
75			if len(self.__conn.response) == 0:
76			raise InvalidUserException()
77
78			if full_dict:
79			return self.__conn.response[0]["attributes"]
80
81			return {
82			"umid":self.__conn.response[0]["attributes"]["entityid"],
83			"uniqname":self.__conn.response[0]["attributes"]["uid"][0],
84			"name":self.__conn.response[0]["attributes"]["displayName"][0]
85			}
86
87			def __lookup(self, k, v):
88			return self.__do_lookup(k, v, self.ATTRIBUTES)
89
90			def __detail_lookup(self, k, v):
91			return self.__do_lookup(k, v,
92			self.ATTRIBUTES + self.DETAILS_ATTRIBUTES,
93			full_dict=True,
94			)
95
96			def lookup_umid(self, umid, details=False):
97			if details:
98			return self.__detail_lookup("entityid", umid)
99			else:
100			return self.__lookup("entityid", umid)
101
102			def lookup_uniqname(self, uniqname, details=False):
103			if details:
104			return self.__detail_lookup("uid", uniqname)
105			else:
106			return self.__lookup("uid", uniqname)
107
108
109			class User(account.Account):
110			__tablename__ = 'users'
111			__mapper_args__ = {'polymorphic_identity': 'user'}
112
113			id = Column(Integer, ForeignKey("accounts.id"), primary_key=True)
114			uniqname = Column(String(8), nullable=False, unique=True)
115			umid = Column(String(8), unique=True)
116			_password = Column("password", String(255))
117			_salt = Column("salt", String(255))
118			enabled = Column(Boolean, nullable=False, default=True)
119			archived = Column(Boolean, nullable=False, default=False)
120			role = Column(Enum("user", "serviceaccount", "manager", "administrator", name="user_type"),
121			nullable=False, default="user")
122
123			administrative_events = relationship(event.Event, foreign_keys=[event.Event.user_id], backref="admin")
124			events_deleted = relationship(event.Event, foreign_keys=[event.Event.deleted_user_id], backref="deleted_user")
125			requests = relationship(request.Request, foreign_keys=[request.Request.user_id], backref="user")
126			request_posts = relationship(
127			request_post.RequestPost,
128			foreign_keys=[request_post.RequestPost.user_id],
129			backref="user",
130			)
131
132			__ldap = LDAPLookup()
133
134			def __init__(self, uniqname, umid, name):
135			self.enabled = True
136			self.uniqname = uniqname
137			self.umid = umid
138			self.name = name
139			self.balance = 0.0
140
141			def __str__(self):
142			return "<User: id {}, uniqname {}, umid {}, name {}, balance {}>".\
143			format(self.id, self.uniqname, self.umid, self.name, self.balance)
144
145			@classmethod
146			def from_id(cls, id):
147			return DBSession.query(cls).filter(cls.id == id).one()
148
149			def get_details(self):
150			return self.__ldap.lookup_uniqname(self.uniqname, details=True)
151
152			@classmethod
153			def from_uniqname(cls, uniqname, local_only=False):
154			u = DBSession.query(cls).filter(cls.uniqname == uniqname).first()
155			if not u and not local_only:
156			u = cls(**cls.__ldap.lookup_uniqname(uniqname))
157			DBSession.add(u)
158			return u
159
160			@classmethod
161			def from_umid(cls, umid, create_if_never_seen=False):
162			u = DBSession.query(cls).filter(cls.umid == umid).first()
163			if not u:
164			if create_if_never_seen:
165			u = cls(**cls.__ldap.lookup_umid(umid))
166			DBSession.add(u)
167			utility.new_user_email(u)
168			else:
169			raise InvalidUserException()
170			return u
171
172			@classmethod
173			def from_fuzzy(cls, search_str, any=True):
174			q = DBSession.query(cls)\
175			.filter(or_(
176			cls.uniqname.ilike('%{}%'.format(search_str)),
177			cls.umid.ilike('%{}%'.format(search_str)),
178			cls.name.ilike('%{}%'.format(search_str))
179			))
180			if not any:
181			q = q.filter(cls.enabled)\
182			.filter(cls.archived==False)
183
184			return q.all()
185
186			@classmethod
187			def all(cls):
188			return DBSession.query(cls)\
189			.filter(cls.enabled)\
190			.order_by(cls.name)\
191			.all()
192
193			@classmethod
194			def count(cls):
195			return DBSession.query(func.count(cls.id).label('c'))\
196			.filter(cls.role != 'serviceaccount')\
197			.filter(cls.archived == False)\
198			.filter(cls.enabled == True)\
199			.one().c
200
201			@classmethod
202			def get_admins(cls):
203			return DBSession.query(cls)\
204			.filter(cls.enabled)\
205			.filter(cls.role=='administrator').all()
206
207			@classmethod
208			def get_shame_users(cls):
209			return DBSession.query(cls)\
210			.filter(cls.enabled)\
211			.filter(cls.balance < -5)\
212			.order_by(cls.balance).all()
213
214			@classmethod
215			def get_normal_users(cls):
216			return DBSession.query(cls)\
217			.filter(cls.enabled)\
218			.filter(cls.archived == False)\
219			.order_by(cls.name).all()
220
221			@classmethod
222			def get_archived_users(cls):
223			return DBSession.query(cls)\
224			.filter(cls.enabled)\
225			.filter(cls.archived == True)\
226			.order_by(cls.name).all()
227
228			@classmethod
229			def get_disabled_users(cls):
230			return DBSession.query(cls)\
231			.filter(cls.enabled == False)\
232			.order_by(cls.name).all()
233
234			@classmethod
235			def get_number_new_users(cls, start=None, end=None):
236			r = DBSession.query(cls)\
237			.filter(cls.enabled == True)
238
239			if start:
240			r = r.filter(cls.created_at>=start)
241			if end:
242			r = r.filter(cls.created_at<end)
243
244			return r.count()
245
246			@classmethod
247			def get_users_total(cls):
248			return DBSession.query(func.sum(User.balance).label("total_balance"))\
249			.one().total_balance or Decimal(0.0)
250
251			# Sum the total amount of money in user accounts that we are holding for
252			# users. This is different from just getting the total because it doesn't
253			# count users with negative balances
254			@classmethod
255			def get_amount_held(cls):
256			return DBSession.query(func.sum(User.balance).label("total_balance"))\
257			.filter(User.balance>0)\
258			.one().total_balance or Decimal(0.0)
259
260			@classmethod
261			def get_amount_owed(cls):
262			return DBSession.query(func.sum(User.balance).label("total_balance"))\
263			.filter(User.balance<0)\
264			.one().total_balance or Decimal(0.0)
265
266			@classmethod
267			def get_user_count_cumulative(cls):
268			rows = DBSession.query(cls.created_at)\
269			.order_by(cls.created_at)\
270			.all()
271			return utility.timeseries_cumulative(rows)
272
273			def iterate_recent_items(self, limit=None, allow_duplicates=False):
274			items = set()
275			count = 0
276			for e in self.events:
277			if e.type == 'purchase':
278			for transaction in e.transactions:
279			if transaction.type == 'purchase':
280			for line_item in transaction.subtransactions:
281			if (line_item.item not in items) or allow_duplicates:
282			count += 1
283			if limit is not None and count > limit:
284			return
285			yield line_item.item
286			items.add(line_item.item)
287
288			def __make_salt(self):
289			return binascii.b2a_base64(open("/dev/urandom", "rb").read(32))[:-3].decode("ascii")
290
291			@hybrid_property
292			def password(self):
293			return self._password
294
295			@password.setter
296			def password(self, password):
297			if password == '':
298			# Use this to clear the password so the user can't login
299			self._password = None
300			else:
301			self._salt = self.__make_salt()
302			salted = (self._salt + password).encode('utf-8')
303			self._password = hashlib.sha256(salted).hexdigest()
304
305			def random_password(self):
306			password = ''.join(random.choice(string.ascii_letters + string.digits) for x in range(6))
307			self._salt = self.__make_salt()
308			salted = (self._salt + password).encode('utf-8')
309			self._password = hashlib.sha256(salted).hexdigest()
310			return password
311
312			def check_password(self, cand):
313			if not self._salt:
314			return False
315			salted = (self._salt + cand).encode('utf-8')
316			c = hashlib.sha256(salted).hexdigest()
317			return c == self._password
318
319			@property
320			def has_password(self):
321			return self._password != None
322
323			# Cash deposit limit is now fixed at $2 because we have a bill acceptor
324			@property
325			def deposit_limit(self):
326			return 2.0
327
328
329			def get_user(request):
330			login = authenticated_userid(request)
331			if not login:
332			return None
333			return DBSession.query(User).filter(User.uniqname == login).one()
334
335
336			# This is in a stupid place due to circular input problems
337			@property
338			def __user_from_foreign_key(self):
339			return User.from_id(self.user_id)
340			event.Event.user = __user_from_foreign_key
341
342
343			def groupfinder(userid, request):
344			user = User.from_uniqname(userid)
345			if user.role == "user":
346			return ["user",]
347			elif user.role == "manager":
348			return ["user","manager"]
349			elif user.role == "administrator":
350			return ["user","manager","admin","serviceaccount"]
351			elif user.role == "serviceaccount":
352			return ["serviceaccount"]
353

um-cseg / chez-betty

Push — master ( 1c4fdd...cd839c )

User.get_number_new_users() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like