Python user 模块，User() 实例源码

def get_user(u):
    row = query_db("SELECT * FROM USERS WHERE NAME = ?", [u.name], one=True)
    u = User()
    if row is None:
        return None
    u.id=row[0]
    u.name=row[1]
    u.password=row[2]
    u.longname=row[3]
    u.email=row[4]
    u.isblack=row[5]
    u.isbaron=row[6]
    u.isshown=row[7]
    u.autoblack=row[8]
    u.onlyrfid=row[9]
    u.rfid_id = get_rfid_ids_by_userid(u.id)
    logging.info(u)
    return u

def get_user_by_name(name):
    row = query_db("SELECT * FROM USERS WHERE NAME = ?", [name], one=True)
    u = User()
    if row is None:
        return None
    u.id=row[0]
    u.name=row[1]
    u.password=row[2]
    u.longname=row[3]
    u.email=row[4]
    u.isblack=row[5]
    u.isbaron=row[6]
    u.isshown=row[7]
    u.autoblack=row[8]
    u.onlyrfid=row[9]
    logging.debug(u)
    u.rfid_id = get_rfid_ids_by_userid(u.id)
    return u

def get_user_by_id(id):
    row = query_db("SELECT * FROM USERS WHERE ID = ?", [id], one=True)
    u = User()
    if row is None:
        return None
    u.id=row[0]
    u.name=row[1]
    u.password=row[2]
    u.longname=row[3]
    u.email=row[4]
    u.isblack=row[5]
    u.isbaron=row[6]
    u.isshown=row[7]
    u.autoblack=row[8]
    u.onlyrfid=row[9]
    u.rfid_id = get_rfid_ids_by_userid(u.id)
    logging.debug(u)
    return u

def get_users():
    rows = query_db("SELECT * FROM USERS")
    users = []
    for row in rows:
        u = User()
        u.id=row[0]
        u.name=row[1]
        u.password=row[2]
        u.longname=row[3]
        u.email=row[4]
        u.isblack=row[5]
        u.isbaron=row[6]
        u.isshown=row[7]
        u.autoblack=row[8]
        u.onlyrfid=row[9]
        u.rfid_id = get_rfid_ids_by_userid(u.id)
        users.append(u)
    return users

def manage_users_add():
    if request.method == 'POST':
        u = User()
        error = None
        u.name = request.form['username'].lower()
        if u.name is None:
            error = "Username not unique!"

        if request.form['password1'] == request.form['password2']:
            #u.password = request.form['password1']
            u.password = bcrypt.hashpw(request.form['password1'], bcrypt.gensalt())
        else:
            error = "Passwords do not match!"
        u.longname = request.form['longname']
        u.email = request.form['email']
        u.rfid_id = request.form['rfid_id']

        if error is None:
            add_user(u)
            return render_template('manage_users_add.html', success="User created!", user=get_user_by_name(session.get('name')));

        return render_template('manage_users_add.html', error=error, user=get_user_by_name(session.get('name')))
    return render_template('manage_users_add.html', user=get_user_by_name(session.get('name')))

def send_mass_mail(name=None):
    if request.method == 'POST':
        users = get_users()
        u = get_user_by_name(name)
        message = request.form['message']
        subject = request.form['subject']

        send_emails(message, subject, users)
        #for user in users:
        #    message_parsed = parse_email(message, user, 3) # change ammount of depts here
        #    subject_parsed = parse_email(subject, user, 3) # change ammount of depts here!
        #    send_email(user.email, subject_parsed, message_parsed)

        success = "An alle User werden Rechnungen versendet."
        return render_template('billing.html', users=users, success=success, dept=0, user=get_user_by_name(session.get('name')))

    if request.method == 'GET':
        return render_template('billing_mass_mail.html', user=get_user_by_name(session.get('name')), emailtext=settings.masMessageStandardText, emailsubject=settings.masMessageStandardSubject)

def do_login():
    username = request.forms.get('username')
    password = get_salted_password()

    current_user = user.get_by_username(username)
    logined = current_user and current_user.salted_password == password

    if logined:
        response.set_cookie('ssl_uid', str(current_user.id))
        response.set_cookie('ssl_pw', password)
        return redirect('/')

    return template('login', 
        username=username, 
        message='User not found.' if not current_user else 'Password is incorrect.', 
        salt=config.USER_SALT
    )

def input_and_create_user(self):
        print("Please input username!")
        users.append(user.User(input()))

def input_and_create_user(self):
        """
        Ask for new username and add it to the list of users.
        """
        print("Please input username!")
        new_username = input()
        new_user = user.User(new_username)
        self.users.append(new_user)

def prompt_user_selection(self):
        """
        Print and select a user from already added users.
        :return: User object which is selected.
        """
        self.print_users()
        user_input = self.input_for_user_selection()
        currently_selected = self.users[int(user_input)]
        return currently_selected

def run(self):  # Run the main program
        print("Welcome to {}!\nLogin at: {}\nType 'help' for manual, 'exit' to terminate the program.".format(self.bank_name, datetime.datetime.now()))  # Print welcome message
        user_input = ""
        while user_input != "exit":  # Start main loop
            # Wait until user input is 'quit', which terminates the program.
            print('==========================================')
            user_input = input()
            # Do actions for input
            try:
                if user_input == CommandLineProgram.ACTION.HELP:
                    # If user command is 'help'
                    print("'help': This command.\n'exit': Close all session and terminate.\n'add user': Add a new user to the session.\n'list users': List all users.\n'add transaction': Add transaction to an existing user.'generate report': Print report for an existing user.")
                elif user_input == CommandLineProgram.ACTION.ADD_USER:
                    # If user command is 'add user'
                    users.append(user.User(input("Please input username!")))
                elif user_input == CommandLineProgram.ACTION.LIST_USERS:
                    # If user command is 'list users'
                    i = 0
                    for item in users:
                        print("{}. {}".format(i, item.name))
                        i = i + 1
                elif user_input == CommandLineProgram.ACTION.ADD_TRANSACTION:
                    # If user command is 'add transaction'
                    self.prompt_user_selection().add_transaction(Transaction(input("Amount of transaction:")))
                elif user_input == CommandLineProgram.ACTION.GENERATE_REPORT:
                    # If user command is 'print report'
                    self.print_all_transaction(self.prompt_user_selection())
            except Exception:
                print("Try again")

def input_and_create_user(self):
        """
        Ask for new username and add it to the list of users.
        """
        print("Please input username!")
        users.append(user.User(input()))

def set_sender(self, sender_username):
        self.sender = User(sender_username)

def set_receiver(self, receiver_username):
        # update only if previous is blank (other case it will be set in parse_message)
        if self.receiver is None:
            self.receiver = User(receiver_username)

def create_from_array(self, arr_tip):
        # import user
        self.receiver = User(arr_tip['receiver'])
        self.sender = User(arr_tip['sender'])
        del arr_tip['receiver']
        del arr_tip['sender']

        for key in arr_tip.keys():
            setattr(self, key, arr_tip[key])

        return self

def initialize(self):
        """Performs initialization that needs to happen after the Bot object is
        constructed.
        """
        self.user = user.User(self,
                              groups.Groups(self.database),
                              steam.client.SteamClient()
        )

def check_user_in_list(id):
    if id not in users_rep:
        current_user = User(id)
        users_rep[id] = current_user
        return current_user

def __init__(self, recipient_id):
        self.user = User(recipient_id)
        self.state = State.ONBOARDING
        self.realtime_subs = Subscriptions()
        self.interest_subs = Subscriptions()
        self.messages = []
        database.Conversations[recipient_id] = self
        database.Users[recipient_id] = self.user

def startElement(self, name, attrs, connection):
        if name == 'Initiator':
            self.initiator = user.User(self)
            return self.initiator
        elif name == 'Owner':
            self.owner = user.User(self)
            return self.owner
        elif name == 'Part':
            part = Part(self.bucket)
            self._parts.append(part)
            return part
        return None

def startElement(self, name, attrs, connection):
        if name == 'Initiator':
            self.initiator = user.User(self)
            return self.initiator
        elif name == 'Owner':
            self.owner = user.User(self)
            return self.owner
        elif name == 'Part':
            part = Part(self.bucket)
            self._parts.append(part)
            return part
        return None

def generate_test_users():

    for i in range(1, 30):
        test = User
        test.name = 'test' + str(i)
        test.email = 'test' + str(i) + '@test.at'
        test.password = 'test'
        test.longname = 'Test User' + str(i)
        test.rfid_id = '0x00000000'
        add_user(test)
    return

def requires_baron(fn):
    @wraps(fn)
    def decorated_function(*a,**kw):
        u = User()
        u.name = session.get('name', None)
        u = get_user(u)
        if not u or not u.isbaron:
            return render_template("not_baron.html", user=get_user_by_name(session.get('name'))), 401
        return fn(*a, **kw)
    return decorated_function

def check_user(u):
    '''
    Check if one user shall be suspended. Returning the reason why suspend, or None.
    @type u: user.User
    '''
    if u.suspended: return None

    msg = None

    for r in u.get_meta("limit", []):
        if not 'type'   in r : continue
        if not 'amount' in r : continue
        if not 'since'  in r : continue 
        type, amount, since = r['type'], r['amount'], r['since']

        now   = datetime.datetime.now()

        since = datetime.datetime(now.year, now.month, 1)                       if since == 'this-month'  else \
                datetime.datetime(now.year, now.month, now.day - now.weekday()) if since == 'this-week'   else \
                datetime.datetime.strptime(since, DATE_FORMAT)

        if   type == 'time':     # Time-to-expire rule. amount: useless
            if now >= since:
                msg = "Expired: %s" % r['since']
                break
        elif type == 'traffic':  # Traffic-limited rule. amount: traffic in bytes.
            tq = traffic.query(uid=u.id, min_time=since.strftime(DATE_FORMAT), sum=traffic.QS_ALL)
            if tq[0][2] > amount:
                msg = "Traffic: used %s, limit %s" % (utils.sizeof_fmt(tq[0][2]), utils.sizeof_fmt(amount))
                break

    return msg

def admin_user_add():
    u = user.User()
    u.username, u.salted_password, u.sskey = [request.forms.get(n) for n in ('username', 'password', 'sskey')]
    u.create()
    u.write()
    return { "id": u.id }

def startElement(self, name, attrs, connection):
        if name == 'Initiator':
            self.initiator = user.User(self)
            return self.initiator
        elif name == 'Owner':
            self.owner = user.User(self)
            return self.owner
        elif name == 'Part':
            part = Part(self.bucket)
            self._parts.append(part)
            return part
        return None

def author_info(self, topic):
        author_url = home_page_url+"/api/members/show.json?username="+topic.author
        response = session.get(author_url)
        data = json.loads(response.content)
        user = User()
        user.id = data.get('id')
        user.name = data.get('username')
        user.website = data.get('website')
        user.twitter = data.get('twitter')
        user.github = data.get('github')
        user.location = data.get('location')
        user.tagline = data.get('tagline')
        user.bio = data.get('bio')
        user.time = format_time(data.get('created'))
        return user

def __init__(self, name: str, owner: User, default=False):
        self.name = name
        self.owner = owner
        self.default = default
        self.users = {}
        self.blocked = {}

def add_user(self, user: User):
        self.users[user.alias] = user

def rem_user(self, user: User):
        self.users.pop(user.alias, None)

def block_user(self, user: User):
        self.blocked[user.alias] = user

def unblock_user(self, user: User):
        self.blocked.pop(user.alias, None)

def set_rating(self, user_or_ip, rating):
        '''Record a user's rating of this package.

        The caller function is responsible for doing the commit.

        If a rating is outside the range MAX_RATING - MIN_RATING then a
        RatingValueException is raised.

        @param user_or_ip - user object or an IP address string
        '''
        user = None
        from user import User
        from rating import Rating, MAX_RATING, MIN_RATING
        if isinstance(user_or_ip, User):
            user = user_or_ip
            rating_query = meta.Session.query(Rating)\
                               .filter_by(package=self, user=user)
        else:
            ip = user_or_ip
            rating_query = meta.Session.query(Rating)\
                               .filter_by(package=self, user_ip_address=ip)

        try:
            rating = float(rating)
        except TypeError:
            raise RatingValueException
        except ValueError:
            raise RatingValueException
        if rating > MAX_RATING or rating < MIN_RATING:
            raise RatingValueException

        if rating_query.count():
            rating_obj = rating_query.first()
            rating_obj.rating = rating
        elif user:
            rating = Rating(package=self,
                            user=user,
                            rating=rating)
            meta.Session.add(rating)
        else:
            rating = Rating(package=self,
                            user_ip_address=ip,
                            rating=rating)
            meta.Session.add(rating)

def manage_users_edit(name=None):
    if request.method == 'GET':
        error = None
        u = User()
        u.name = name
        u = get_user(u)

        if u is None:
            error = "User existiert nicht"

        return render_template('manage_users_edit.html', user_to_edit=u, error=error, user=get_user_by_name(session.get('name')))

    if request.method == 'POST':
        u = User()
        #print request.form
        u.id = request.form['id']
        u.name = request.form['username'].lower()
        u.longname=request.form['longname']
        u.email = request.form['email']
        u.rfid_id = request.form['rfid_id']

        if len(request.form['new_password']) > 0:
            u.password = bcrypt.hashpw(request.form['new_password'], bcrypt.gensalt())
        else:
            u2 = get_user_by_id(u.id)
            u.password = u2.password

        if 'isblack' in request.form:
            u.isblack = True
        else:
            u.isblack = False

        if 'isbaron' in request.form:
            u.isbaron = True
        else:
            u.isbaron = False

        if 'isshown' in request.form:
            u.isshown = True
        else:
            u.isshown = False

        if 'autoblack' in request.form:
            u.autoblack = True
        else:
            u.autoblack = False

        if 'onlyrfid' in request.form:
            u.onlyrfid = True
        else:
            u.onlyrfid = False

        update_user(u)

        return redirect('/manage_users')