calculate-utils-4-lib/calculate/server/utils/users.py

from fastapi import HTTPException, status
from sqlalchemy import func, and_
from typing import List

from ..database.db import database
from ..database.users import users_table as ut, users_rights, rights_table

from ..schemas.users import UserCreate, User


async def get_user_by_username(username: str):
    '''Метод для получения строки с данными пользователя из базы данных по
    username.'''
    query = ut.select().where(ut.c.login == username)
    user_data = await database.fetch_one(query)
    query = (ut.
             join(users_rights).
             join(rights_table).
             select().
             where(and_(ut.c.id == users_rights.c.user_id,
                        ut.c.login == username,
                        rights_table.c.id == users_rights.c.right_id)).
             with_only_columns([ut.c.id,
                                ut.c.login,
                                ut.c.password,
                                func.group_concat(rights_table.c.name,
                                                  ' ').label("rights")]).
             group_by(ut.c.id))
    response = await database.fetch_one(query)

    user_data = dict(response)
    user_data['rights'] = user_data['rights'].split()
    return user_data


async def create_user(username: str, hashed_password: str, rights: List[str]):
    UserCreate(login=username,
               password=hashed_password,
               rights=rights)


def check_user_rights(user_data: User, *rights: List[str]) -> None:
    user_rights = user_data.rights
    for right in user_rights:
        if right not in user_rights:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
                                detail='Not enough permissions',
                                headers={"WWW-Authenticate": "Bearer"}
                                )