<commit_before># stdlib
from typing import Any
from typing import List
from typing import Optional
from typing import Tuple
from typing import Union

# third party
from bcrypt import checkpw
from bcrypt import gensalt
from bcrypt import hashpw
from nacl.encoding import HexEncoder
from nacl.signing import SigningKey
from nacl.signing import VerifyKey
from pydantic import BaseModel
from pydantic import EmailStr
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Query
from sqlalchemy.orm import sessionmaker

# relative
from ..exceptions import InvalidCredentialsError
from ..exceptions import UserNotFoundError
from ..node_table.pdf import PDFObject
from ..node_table.roles import Role
from ..node_table.user import SyftUser
from ..node_table.user import UserApplication
from .database_manager import DatabaseManager
from .role_manager import RoleManager


# Shared properties
class UserBase(BaseModel):
    email: Optional[EmailStr] = None
    is_active: Optional[bool] = True
    is_superuser: bool = False
    full_name: Optional[str] = None


# Properties to receive via API on creation
class UserCreate(UserBase):
    email: EmailStr
    password: str


# Properties to receive via API on update
class UserUpdate(UserBase):
    password: Optional[str] = None


class UserInDBBase(UserBase):
    id: Optional[int] = None

    class Config:
        orm_mode = True


# Additional properties to return via API
class User(UserInDBBase):
    pass


# Additional properties stored in DB
class UserInDB(UserInDBBase):
    hashed_password: str


class UserManager(DatabaseManager):

    schema = SyftUser

    def __init__(self, database: Engine) -> None:
        super().__init__(schema=UserManager.schema, db=database)
        self.roles = RoleManager(database)

    @property
    def common_users(self) -> list:
        common_users: List[SyftUser] = []
        for role in self.roles.common_roles:
            common_users = common_users + list(super().query(role=role.id))

        return common_users

    @property
    def org_users(self) -> list:
        org_users: List[SyftUser] = []
        for role in self.roles.org_roles:
            org_users = org_users + list(super().query(role=role.id))
        return org_users

    def create_user_application(
        self,
        name: str,
        email: str,
        password: str,
        daa_pdf: Optional[bytes],
        institution: Optional[str] = "",
        website: Optional[str] = "",
    ) -> None:
        salt, hashed = self.__salt_and_hash_password(password, 12)
        session_local = sessionmaker(autocommit=False, autoflush=False, bind=self.db)()
        _pdf_obj = PDFObject(binary=daa_pdf)
        session_local.add(_pdf_obj)
        session_local.commit()
        session_local.flush()
        session_local.refresh(_pdf_obj)

        _obj = UserApplication(
            name=name,
            email=email,
            salt=salt,
            hashed_password=<PASSWORD>,
            daa_pdf=_pdf_obj.id,
            institution=institution,
            website=website,
        )
        session_local.add(_pdf_obj)
        session_local.add(_obj)
        session_local.commit()

    def get_all_applicant(self) -> List[UserApplication]:
        session_local = sessionmaker(autocommit=False, autoflush=False, bind=self.db)()
        result = list(session_local.query(UserApplication).all())
        session_local.close()
        return result

    def process_user_application(
        self, candidate_id: int, status: str, verify_key: VerifyKey
    ) -> None:
        session_local = sessionmaker(autocommit=False, autoflush=False, bind=self.db)()
        candidate = (
            session_local.query(UserApplication).filter_by(id=candidate_id).first()
        )
        session_local.close()

        if status == "accepted":
            # Generate a new signing key
            _private_key = SigningKey.generate()

            encoded_pk = _private_key.encode(encoder=HexEncoder).decode("utf-8")
            encoded_vk = _private_key.verify_key.encode(encoder=HexEncoder).decode(
                "utf-8"
            )
            added_by = self.get_user(verify_key).name
            self.register(
                name=candidate.name,
                email=candidate.email,
                role=self.roles.ds_role.id,
                budget=0,
                private_key=encoded_pk,
                verify_key=encoded_vk,
                hashed_password=<PASSWORD>,
                salt=candidate.salt,
                daa_pdf=candidate.daa_pdf,
                added_by=added_by,
                institution=candidate.institution,
                website=candidate.website,
            )
        else:
            status = "rejected"

        session_local = sessionmaker(autocommit=False, autoflush=False, bind=self.db)()
        candidate = (
            session_local.query(UserApplication).filter_by(id=candidate_id).first()
        )
        candidate.status = status
        session_local.flush()
        session_local.commit()
        session_local.close()

    def signup(
        self,<commit_msg>Save created_at value during user creation<commit_after># stdlib
from typing import Any
from typing import List
from typing import Optional
from typing import Tuple
from typing import Union
from datetime import datetime

# third party
from bcrypt import checkpw
from bcrypt import gensalt
from bcrypt import hashpw
from nacl.encoding import HexEncoder
from nacl.signing import SigningKey
from nacl.signing import VerifyKey
from pydantic import BaseModel
from pydantic import EmailStr
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Query
from sqlalchemy.orm import sessionmaker

# relative
from ..exceptions import InvalidCredentialsError
from ..exceptions import UserNotFoundError
from ..node_table.pdf import PDFObject
from ..node_table.roles import Role
from ..node_table.user import SyftUser
from ..node_table.user import UserApplication
from .database_manager import DatabaseManager
from .role_manager import RoleManager


# Shared properties
class UserBase(BaseModel):
    email: Optional[EmailStr] = None
    is_active: Optional[bool] = True
    is_superuser: bool = False
    full_name: Optional[str] = None


# Properties to receive via API on creation
class UserCreate(UserBase):
    email: EmailStr
    password: str


# Properties to receive via API on update
class UserUpdate(UserBase):
    password: Optional[str] = None


class UserInDBBase(UserBase):
    id: Optional[int] = None

    class Config:
        orm_mode = True


# Additional properties to return via API
class User(UserInDBBase):
    pass


# Additional properties stored in DB
class UserInDB(UserInDBBase):
    hashed_password: str


class UserManager(DatabaseManager):

    schema = SyftUser

    def __init__(self, database: Engine) -> None:
        super().__init__(schema=UserManager.schema, db=database)
        self.roles = RoleManager(database)

    @property
    def common_users(self) -> list:
        common_users: List[SyftUser] = []
        for role in self.roles.common_roles:
            common_users = common_users + list(super().query(role=role.id))

        return common_users

    @property
    def org_users(self) -> list:
        org_users: List[SyftUser] = []
        for role in self.roles.org_roles:
            org_users = org_users + list(super().query(role=role.id))
        return org_users

    def create_user_application(
        self,
        name: str,
        email: str,
        password: str,
        daa_pdf: Optional[bytes],
        institution: Optional[str] = "",
        website: Optional[str] = "",
    ) -> None:
        salt, hashed = self.__salt_and_hash_password(password, 12)
        session_local = sessionmaker(autocommit=False, autoflush=False, bind=self.db)()
        _pdf_obj = PDFObject(binary=daa_pdf)
        session_local.add(_pdf_obj)
        session_local.commit()
        session_local.flush()
        session_local.refresh(_pdf_obj)

        _obj = UserApplication(
            name=name,
            email=email,
            salt=salt,
            hashed_password=hashed,
            daa_pdf=_pdf_obj.id,
            institution=institution,
            website=website,
        )
        session_local.add(_pdf_obj)
        session_local.add(_obj)
        session_local.commit()

    def get_all_applicant(self) -> List[UserApplication]:
        session_local = sessionmaker(autocommit=False, autoflush=False, bind=self.db)()
        result = list(session_local.query(UserApplication).all())
        session_local.close()
        return result

    def process_user_application(
        self, candidate_id: int, status: str, verify_key: VerifyKey
    ) -> None:
        session_local = sessionmaker(autocommit=False, autoflush=False, bind=self.db)()
        candidate = (
            session_local.query(UserApplication).filter_by(id=candidate_id).first()
        )
        session_local.close()

        if status == "accepted":
            # Generate a new signing key
            _private_key = SigningKey.generate()

            encoded_pk = _private_key.encode(encoder=HexEncoder).decode("utf-8")
            encoded_vk = _private_key.verify_key.encode(encoder=HexEncoder).decode(
                "utf-8"
            )
            added_by = self.get_user(verify_key).name
            self.register(
                name=candidate.name,
                email=candidate.email,
                role=self.roles.ds_role.id,
                budget=0,
                private_key=encoded_pk,
                verify_key=encoded_vk,
                hashed_password=<PASSWORD>,
                salt=candidate.salt,
                daa_pdf=candidate.daa_pdf,
                added_by=added_by,
                institution=candidate.institution,
                website=candidate.website,
                created_at=datetime.now(),
            )
        else:
            status = "rejected"

        session_local = sessionmaker(autocommit=False, autoflush=False, bind=self.db)()
        candidate = (
            session_local.query(UserApplication).filter_by(id=candidate_id).first()
        )
        candidate.status = status
        session_local.flush()
        session_local.commit()
        session_local.close()

    def signup(
        self,
{'fraction_non_alphanumeric': 0.08428709990300678, 'fraction_numerical': 0.0009699321047526673, 'mean_word_length': 2.5851877607788594, 'pattern_counts': {'":': 2, '<': 6, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1373649', 'n_tokens_mistral': 2979, 'n_tokens_neox': 2891, 'n_words': 669}