latosa/user.py

import os

from flask import Flask
from flask_bcrypt import Bcrypt

from db import DB


class User:

    def __init__(self,
                 app: Flask,
                 uid: str,
                 display_name: str,
                 email: str,
                 password: str,
                 salt: str,
                 admin: bool = False,
                 groups: list[str] = []):
        self.db = DB()
        self.display_name = display_name
        self.email = email
        self.groups = groups
        self.is_active = False
        self.is_admin = admin
        self.is_anonymous = False
        self.is_authenticated = False
        self.uid = uid
        self.bcrypt = Bcrypt(app)
        self.salt = salt
        self.password_hash = self.bcrypt.generate_password_hash(
            password + self.salt).decode('utf-8')
        self.commit()

    def check_password(self, password: str):
        return self.bcrypt.check_password_hash(self.password_hash,
                                               password + self.salt)

    def get_id(self):
        return self.uid

    def get_display_name(self):
        return self.display_name

    def get_email(self):
        return self.email

    def get_groups(self):
        return ','.join(self.groups)

    def set_active(self, active: bool):
        self.is_active = active
        self.commit()

    def set_authenticated(self, authenticated: bool):
        self.is_authenticated = authenticated
        self.commit()

    def set_anonymous(self, anonymous: bool):
        self.is_anonymous = anonymous
        self.commit()

    def set_admin(self, admin: bool):
        self.is_admin = admin
        self.commit()

    def set_email(self, email: str):
        self.email = email
        self.commit()

    def set_password(self, password: str):
        self.password_hash = self.bcrypt.generate_password_hash(
            password + self.salt).decode('utf-8')
        self.commit()

    def commit(self):
        bind_params = {
            'uid': self.uid,
            'display_name': self.display_name,
            'is_active': self.is_active,
            'is_anonymous': self.is_anonymous,
            'is_admin': self.is_admin,
            'email': self.email,
            'password_hash': self.password_hash,
            'salt': self.salt,
            'groups': self.get_groups(),
        }
        statement = "INSERT OR REPLACE INTO users (uid, display_name, is_active, is_anonymous, is_admin, email, password_hash, salt, groups) VALUES(:uid, :display_name, :is_active, :is_anonymous, :is_admin, :email, :password_hash, :salt, :groups)"
        self.db.execute(statement, bind_params)

    @staticmethod
    def users_from_db() -> list:
        db = DB()
        statement = "SELECT * FROM users"
        result = db.execute(statement, {})
        users = []
        for row in result:
            users.append(User(**row))
        return users