|
|
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"}
|
|
|
)
|
