first commit
This commit is contained in:
@@ -0,0 +1,80 @@
|
||||
from sqlalchemy import create_engine, select
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from src.model.database_model.model import User, Permissions
|
||||
|
||||
class ActionsDB:
|
||||
def __init__(self) -> None:
|
||||
self.engine=create_engine("sqlite:///DB/database.db", echo=True)
|
||||
self.Session=sessionmaker(self.engine,expire_on_commit=False)
|
||||
|
||||
|
||||
def get_user_by_email(self, email: str) -> User | None:
|
||||
with self.Session() as session:
|
||||
with session.begin():
|
||||
q=select(User).where(User.email==email)
|
||||
r=session.scalars(q).first()
|
||||
if r:
|
||||
return r
|
||||
else:
|
||||
return None
|
||||
|
||||
def get_user_by_id(self, id:int)->User|None:
|
||||
with self.Session() as session:
|
||||
with session.begin():
|
||||
q=select(User).where(User.id==id)
|
||||
r=session.scalars(q).first()
|
||||
if r:
|
||||
return r
|
||||
else:
|
||||
return None
|
||||
|
||||
def create_user(self, user:dict, permission_names:list[str])->User:
|
||||
with self.Session() as session:
|
||||
with session.begin():
|
||||
new_user=User(**user)
|
||||
if permission_names:
|
||||
perms = session.scalars(
|
||||
select(Permissions).where(Permissions.permission.in_(permission_names))
|
||||
).all()
|
||||
new_user.permissions.extend(perms)
|
||||
session.add(new_user)
|
||||
session.refresh(new_user, attribute_names=["permissions"])
|
||||
return new_user
|
||||
|
||||
def delete_user(self, id:int)->bool:
|
||||
with self.Session() as session:
|
||||
with session.begin():
|
||||
q=select(User).where(User.id==id)
|
||||
r=session.scalar(q)
|
||||
if r is None:
|
||||
return False
|
||||
r.status = False
|
||||
return True
|
||||
|
||||
def update_user(self, data: dict, permission_names: list[str]) -> User | None:
|
||||
with self.Session() as session:
|
||||
with session.begin():
|
||||
q = select(User).where(User.id == data["id"])
|
||||
user = session.execute(q).scalar_one_or_none()
|
||||
if user is None:
|
||||
return None
|
||||
|
||||
if permission_names:
|
||||
perms = session.scalars(
|
||||
select(Permissions).where(Permissions.permission.in_(permission_names))
|
||||
).all()
|
||||
user.permissions = list(perms)
|
||||
|
||||
for key, value in data.items():
|
||||
if key != "id":
|
||||
setattr(user, key, value)
|
||||
|
||||
session.refresh(user, attribute_names=["permissions"])
|
||||
return user
|
||||
|
||||
def get_all_permissions(self)->list[str]:
|
||||
with self.Session() as session:
|
||||
with session.begin():
|
||||
q=select(Permissions)
|
||||
permissions=session.scalars(q).all()
|
||||
return [p.permission for p in permissions]
|
||||
@@ -0,0 +1,17 @@
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
|
||||
class Errors:
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
def credentials_error(self, detail:str)->HTTPException:
|
||||
raise HTTPException(status_code=401, detail=detail, headers={"WWW-Authenticate":"Bearer"})
|
||||
|
||||
def forbidden_error(self,detail:str)->HTTPException:
|
||||
raise HTTPException(status_code=403, detail=detail, headers={"Cache-Control": "no-store, max-age=0"})
|
||||
|
||||
def not_found_error(self, detail:str)->HTTPException:
|
||||
raise HTTPException(status_code=404, detail=detail, headers={"Cache-Control": "no-store, max-age=0"})
|
||||
errors=Errors()
|
||||
@@ -0,0 +1 @@
|
||||
Generic single-database configuration.
|
||||
@@ -0,0 +1,82 @@
|
||||
from logging.config import fileConfig
|
||||
|
||||
from src.db.database import engine
|
||||
from src.model.database_model.model import Model
|
||||
import src.model.database_model.model as models
|
||||
from sqlalchemy import engine_from_config
|
||||
from sqlalchemy import pool
|
||||
|
||||
from alembic import context
|
||||
|
||||
# this is the Alembic Config object, which provides
|
||||
# access to the values within the .ini file in use.
|
||||
config = context.config
|
||||
|
||||
# Interpret the config file for Python logging.
|
||||
# This line sets up loggers basically.
|
||||
if config.config_file_name is not None:
|
||||
fileConfig(config.config_file_name)
|
||||
|
||||
# add your model's MetaData object here
|
||||
# for 'autogenerate' support
|
||||
# from myapp import mymodel
|
||||
# target_metadata = mymodel.Base.metadata
|
||||
target_metadata = Model.metadata
|
||||
config.set_main_option("sqlalchemy.url", engine.url.render_as_string(hide_password=False))
|
||||
|
||||
# other values from the config, defined by the needs of env.py,
|
||||
# can be acquired:
|
||||
# my_important_option = config.get_main_option("my_important_option")
|
||||
# ... etc.
|
||||
|
||||
|
||||
def run_migrations_offline() -> None:
|
||||
"""Run migrations in 'offline' mode.
|
||||
|
||||
This configures the context with just a URL
|
||||
and not an Engine, though an Engine is acceptable
|
||||
here as well. By skipping the Engine creation
|
||||
we don't even need a DBAPI to be available.
|
||||
|
||||
Calls to context.execute() here emit the given string to the
|
||||
script output.
|
||||
|
||||
"""
|
||||
url = config.get_main_option("sqlalchemy.url")
|
||||
context.configure(
|
||||
url=url,
|
||||
target_metadata=target_metadata,
|
||||
literal_binds=True,
|
||||
dialect_opts={"paramstyle": "named"},
|
||||
)
|
||||
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
|
||||
|
||||
def run_migrations_online() -> None:
|
||||
"""Run migrations in 'online' mode.
|
||||
|
||||
In this scenario we need to create an Engine
|
||||
and associate a connection with the context.
|
||||
|
||||
"""
|
||||
connectable = engine_from_config(
|
||||
config.get_section(config.config_ini_section, {}),
|
||||
prefix="sqlalchemy.",
|
||||
poolclass=pool.NullPool,
|
||||
)
|
||||
|
||||
with connectable.connect() as connection:
|
||||
context.configure(
|
||||
connection=connection, target_metadata=target_metadata, render_as_batch=True
|
||||
)
|
||||
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
|
||||
|
||||
if context.is_offline_mode():
|
||||
run_migrations_offline()
|
||||
else:
|
||||
run_migrations_online()
|
||||
@@ -0,0 +1,28 @@
|
||||
"""${message}
|
||||
|
||||
Revision ID: ${up_revision}
|
||||
Revises: ${down_revision | comma,n}
|
||||
Create Date: ${create_date}
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
${imports if imports else ""}
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = ${repr(up_revision)}
|
||||
down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)}
|
||||
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
|
||||
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Upgrade schema."""
|
||||
${upgrades if upgrades else "pass"}
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Downgrade schema."""
|
||||
${downgrades if downgrades else "pass"}
|
||||
@@ -0,0 +1,74 @@
|
||||
"""0.1.0
|
||||
|
||||
Revision ID: ea8d55c2c0a6
|
||||
Revises:
|
||||
Create Date: 2026-07-10 17:19:16.017750
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = 'ea8d55c2c0a6'
|
||||
down_revision: Union[str, Sequence[str], None] = None
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Upgrade schema."""
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.create_table('permissions',
|
||||
sa.Column('id', sa.Integer(), nullable=False),
|
||||
sa.Column('permission', sa.String(length=255), nullable=False),
|
||||
sa.PrimaryKeyConstraint('id', name=op.f('pk_permissions')),
|
||||
sa.UniqueConstraint('permission', name=op.f('uq_permissions_permission'))
|
||||
)
|
||||
with op.batch_alter_table('permissions', schema=None) as batch_op:
|
||||
batch_op.create_index(batch_op.f('ix_permissions_id'), ['id'], unique=False)
|
||||
|
||||
op.create_table('users',
|
||||
sa.Column('id', sa.Integer(), nullable=False),
|
||||
sa.Column('name', sa.String(length=64), nullable=False),
|
||||
sa.Column('last_name', sa.String(length=64), nullable=False),
|
||||
sa.Column('middle_name', sa.String(length=64), nullable=False),
|
||||
sa.Column('email', sa.String(length=255), nullable=False),
|
||||
sa.Column('status', sa.Boolean(), nullable=False),
|
||||
sa.Column('hashed_password', sa.String(length=255), nullable=False),
|
||||
sa.PrimaryKeyConstraint('id', name=op.f('pk_users'))
|
||||
)
|
||||
with op.batch_alter_table('users', schema=None) as batch_op:
|
||||
batch_op.create_index(batch_op.f('ix_users_email'), ['email'], unique=True)
|
||||
batch_op.create_index(batch_op.f('ix_users_last_name'), ['last_name'], unique=False)
|
||||
batch_op.create_index(batch_op.f('ix_users_middle_name'), ['middle_name'], unique=False)
|
||||
batch_op.create_index(batch_op.f('ix_users_name'), ['name'], unique=False)
|
||||
|
||||
op.create_table('user_permission',
|
||||
sa.Column('user_id', sa.Integer(), nullable=False),
|
||||
sa.Column('permission_id', sa.Integer(), nullable=False),
|
||||
sa.ForeignKeyConstraint(['permission_id'], ['permissions.id'], name=op.f('fk_user_permission_permission_id_permissions')),
|
||||
sa.ForeignKeyConstraint(['user_id'], ['users.id'], name=op.f('fk_user_permission_user_id_users')),
|
||||
sa.PrimaryKeyConstraint('user_id', 'permission_id', name=op.f('pk_user_permission'))
|
||||
)
|
||||
# ### end Alembic commands ###
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Downgrade schema."""
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.drop_table('user_permission')
|
||||
with op.batch_alter_table('users', schema=None) as batch_op:
|
||||
batch_op.drop_index(batch_op.f('ix_users_name'))
|
||||
batch_op.drop_index(batch_op.f('ix_users_middle_name'))
|
||||
batch_op.drop_index(batch_op.f('ix_users_last_name'))
|
||||
batch_op.drop_index(batch_op.f('ix_users_email'))
|
||||
|
||||
op.drop_table('users')
|
||||
with op.batch_alter_table('permissions', schema=None) as batch_op:
|
||||
batch_op.drop_index(batch_op.f('ix_permissions_id'))
|
||||
|
||||
op.drop_table('permissions')
|
||||
# ### end Alembic commands ###
|
||||
@@ -0,0 +1,47 @@
|
||||
from sqlalchemy import Boolean, MetaData, create_engine, String, ForeignKey, Table, Column
|
||||
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column,relationship
|
||||
|
||||
|
||||
engine = create_engine("sqlite:///DB/database.db", echo=True)
|
||||
|
||||
'''remember as a boilerplate, or just cp/pst'''
|
||||
class Model(DeclarativeBase):
|
||||
metadata = MetaData(naming_convention={
|
||||
"ix": "ix_%(column_0_label)s",
|
||||
"uq": "uq_%(table_name)s_%(column_0_name)s",
|
||||
"ck": "ck_%(table_name)s_%(constraint_name)s",
|
||||
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
|
||||
"pk": "pk_%(table_name)s",
|
||||
})
|
||||
|
||||
class User(Model):
|
||||
__tablename__ = "users"
|
||||
|
||||
id:Mapped[int] = mapped_column(primary_key=True)
|
||||
name:Mapped[str] = mapped_column(String(64), index=True)
|
||||
last_name:Mapped[str]=mapped_column(String(64), index=True)
|
||||
middle_name:Mapped[str]=mapped_column(String(64), index=True)
|
||||
email:Mapped[str]=mapped_column(String(255), index=True, unique=True)
|
||||
status:Mapped[bool]=mapped_column(Boolean)
|
||||
hashed_password:Mapped[str]=mapped_column(String(255))
|
||||
|
||||
permissions:Mapped[list["Permissions"]]=relationship(secondary="user_permission", back_populates="user", lazy="selectin")
|
||||
def __repr__(self) -> str:
|
||||
return f"ID: {self.id}, Name: {self.name}, Status: {self.status}"
|
||||
|
||||
user_permission = Table(
|
||||
"user_permission",
|
||||
Model.metadata,
|
||||
Column("user_id", ForeignKey("users.id"), primary_key=True),
|
||||
Column("permission_id", ForeignKey("permissions.id"), primary_key=True),
|
||||
)
|
||||
|
||||
class Permissions(Model):
|
||||
__tablename__ = "permissions"
|
||||
|
||||
id:Mapped[int]=mapped_column(primary_key=True, index=True)
|
||||
permission:Mapped[str]=mapped_column(String(255), unique=True)
|
||||
|
||||
user:Mapped[list['User']]=relationship(secondary="user_permission", back_populates="permissions")
|
||||
def __repr__(self)->str:
|
||||
return f"ID: {self.id}, Permissions: {self.permission}"
|
||||
@@ -0,0 +1,12 @@
|
||||
from pydantic_settings import SettingsConfigDict, BaseSettings
|
||||
|
||||
class Settings(BaseSettings):
|
||||
ALGORITHM:str
|
||||
EXPIRE_IN_MINUTES:int
|
||||
SECRET_STRING:str
|
||||
|
||||
model_config=SettingsConfigDict(
|
||||
env_file=".env",
|
||||
extra="ignore"
|
||||
)
|
||||
ENV=Settings()
|
||||
@@ -0,0 +1,50 @@
|
||||
from typing import Annotated
|
||||
from pydantic import BaseModel, EmailStr, Field
|
||||
|
||||
|
||||
|
||||
|
||||
class Base(BaseModel):
|
||||
pass
|
||||
|
||||
|
||||
class PermissionOut(Base):
|
||||
id: int
|
||||
permission: str
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
class PermissionIn(Base):
|
||||
permission: str
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
class UserOut(Base):
|
||||
id:int=Field(..., description="Id of the user")
|
||||
name:str=Field(...,max_length=64, description="name of the user")
|
||||
last_name:str=Field(...,max_length=64, description="Last name of the user")
|
||||
middle_name:str = Field(..., max_length=64, description="middle name of the user")
|
||||
email:EmailStr = Field(...,max_length=255, description="email of the user")
|
||||
status:bool = Field(..., description="status of the user")
|
||||
permissions:list[PermissionOut] = Field(..., description="permissions of the user")
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
class UserCreate(Base):
|
||||
|
||||
name:str=Field(...,max_length=64, description="name of the user")
|
||||
last_name:str=Field(...,max_length=64,description="Last name of the user")
|
||||
middle_name:str = Field(...,max_length=64, description="middle name of the user")
|
||||
email:EmailStr = Field(...,max_length=255, min_length=5, description="email of the user")
|
||||
plain_password:str=Field(...,min_length=8, max_length=72, description="plain password of the user")
|
||||
permissions:list[PermissionIn] = Field(..., description="permissions of the user")
|
||||
|
||||
class UserUpdate(Base):
|
||||
name:Annotated[str|None, Field(None,max_length=64, description="name of the user")]
|
||||
last_name:Annotated[str|None, Field(None,max_length=64,description="Last name of the user")]
|
||||
middle_name:Annotated[str|None, Field(None,max_length=64, description="middle name of the user")]
|
||||
email:Annotated[EmailStr|None, Field(None,max_length=255, min_length=5, description="email of the user")]
|
||||
plain_password:Annotated[str|None, Field(None,min_length=8, max_length=72, description="plain password of the user")]
|
||||
permissions:Annotated[list[PermissionIn]|None, Field(None, description="permissions of the user")]
|
||||
target_id:Annotated[int|None, Field(None, description="Id of the being updated user")]
|
||||
status:Annotated[bool|None, Field(None,description="status of the account")]
|
||||
@@ -0,0 +1,42 @@
|
||||
import bcrypt
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from jose import JWTError, jwt
|
||||
from src.model.env_read.env import ENV
|
||||
from src.errors.errors import errors
|
||||
|
||||
|
||||
'''JWT Create/Decode'''
|
||||
|
||||
class JWT:
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
def jwt_create(self, data:dict)->str:
|
||||
|
||||
encoded_to=data.copy()
|
||||
encoded_to.update({"exp":datetime.now(timezone.utc)+timedelta(minutes=ENV.EXPIRE_IN_MINUTES)})
|
||||
return jwt.encode(encoded_to, ENV.SECRET_STRING, ENV.ALGORITHM)
|
||||
|
||||
def jwt_decode(self, token:str)->dict:
|
||||
|
||||
try:
|
||||
|
||||
payload=jwt.decode(token, ENV.SECRET_STRING, algorithms=[ENV.ALGORITHM])
|
||||
if ((payload.get("sub")) is None):
|
||||
raise errors.credentials_error(detail="JWT payload is None")
|
||||
return payload
|
||||
except JWTError as e:
|
||||
raise errors.credentials_error(detail="JWTError") from e
|
||||
|
||||
'''hash/check'''
|
||||
|
||||
class Hash:
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
def plain_to_hash(self,plain_password:str)->str:
|
||||
return bcrypt.hashpw(plain_password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
||||
|
||||
def hash_check(self,plain_password:str,hashed_password:str)->bool:
|
||||
return bcrypt.checkpw(plain_password.encode("utf-8"),hashed_password.encode("utf-8"))
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
from src.service.JWT.jwt import JWT, Hash
|
||||
from src.db.database import ActionsDB
|
||||
from src.errors.errors import errors
|
||||
|
||||
class Auth:
|
||||
def __init__(self) -> None:
|
||||
|
||||
self.DB=ActionsDB()
|
||||
self.jwt_service = JWT()
|
||||
self.hashes=Hash()
|
||||
|
||||
def create_token(self, form_data_email:str, form_data_password:str)->dict:
|
||||
|
||||
user=self.DB.get_user_by_email(email=form_data_email)
|
||||
|
||||
if not user or not self.hashes.hash_check(form_data_password, user.hashed_password):
|
||||
raise errors.credentials_error(detail="Wrong credentials in create_token")
|
||||
|
||||
if user.status is False:
|
||||
raise errors.credentials_error(detail="This User is deactivated")
|
||||
|
||||
return {"access_token":self.jwt_service.jwt_create({"sub":str(user.id)}), "token_type":"bearer"}
|
||||
|
||||
def get_current_user(self, token:str):
|
||||
|
||||
payload = self.jwt_service.jwt_decode(token=token)
|
||||
sub = payload.get("sub")
|
||||
|
||||
if sub is None:
|
||||
raise errors.credentials_error(detail="JWT token is incorrect")
|
||||
try:
|
||||
user_id = int(sub)
|
||||
except (TypeError, ValueError):
|
||||
raise errors.credentials_error(detail="JWT payload is incorrect")
|
||||
|
||||
user = self.DB.get_user_by_id(id=user_id)
|
||||
if not user:
|
||||
raise errors.credentials_error(detail="JWT tries to allocate a non-existing user")
|
||||
|
||||
return user
|
||||
|
||||
auth=Auth()
|
||||
@@ -0,0 +1,128 @@
|
||||
from src.db.database import ActionsDB
|
||||
from src.service.JWT.jwt import Hash
|
||||
from src.errors.errors import errors
|
||||
from src.model.database_model.model import User
|
||||
from src.model.user.user_model import UserOut, UserCreate, UserUpdate
|
||||
class CrudActions:
|
||||
|
||||
def __init__(self) -> None:
|
||||
|
||||
self.DB = ActionsDB()
|
||||
self.hashes=Hash()
|
||||
|
||||
|
||||
def get_user_by_email(self, email: str, current_user:UserOut)->User:
|
||||
|
||||
allowed = {"can_view", "admin"}
|
||||
|
||||
if not any(p.permission in allowed for p in current_user.permissions):
|
||||
raise errors.forbidden_error(detail="You don't have permissions to get user by email")
|
||||
|
||||
user = self.DB.get_user_by_email(email=email)
|
||||
if user is None:
|
||||
raise errors.not_found_error(detail="User not found")
|
||||
|
||||
return user
|
||||
|
||||
def get_user_by_id(self, id:int, current_user:UserOut)->User:
|
||||
|
||||
allowed = {"can_view", "admin"}
|
||||
|
||||
if not any(p.permission in allowed for p in current_user.permissions):
|
||||
raise errors.forbidden_error(detail="You don't have permissions to get user by id")
|
||||
|
||||
user = self.DB.get_user_by_id(id=id)
|
||||
if user is None:
|
||||
raise errors.not_found_error(detail="User not found")
|
||||
|
||||
return user
|
||||
|
||||
|
||||
def create_user(self, data: UserCreate, current_user: UserOut) -> User:
|
||||
|
||||
allowed = {"admin", "can_create"}
|
||||
|
||||
permission_names = [p.permission for p in data.permissions]
|
||||
is_admin = any(p.permission == "admin" for p in current_user.permissions)
|
||||
existing_permissions = set(self.DB.get_all_permissions())
|
||||
|
||||
user_info = data.model_dump(exclude={"permissions"})
|
||||
|
||||
user_info["status"]=True
|
||||
|
||||
if not any(p.permission in allowed for p in current_user.permissions):
|
||||
raise errors.forbidden_error(detail="You don't have permissions to create users")
|
||||
|
||||
if not set(permission_names).issubset(existing_permissions):
|
||||
raise errors.not_found_error(detail="You have requested the unknown permission")
|
||||
|
||||
if not is_admin and "admin" in permission_names:
|
||||
raise errors.forbidden_error(detail="You can't create user with admin permissions")
|
||||
|
||||
if self.DB.get_user_by_email(email=user_info["email"]):
|
||||
raise errors.forbidden_error(detail="You can't use this email")
|
||||
|
||||
user_info["hashed_password"] = self.hashes.plain_to_hash(user_info.pop("plain_password"))
|
||||
|
||||
user = self.DB.create_user(user=user_info, permission_names=permission_names)
|
||||
|
||||
if user is None:
|
||||
raise errors.not_found_error(detail="User not found")
|
||||
return user
|
||||
|
||||
|
||||
def delete_user(self,id:int, current_user:UserOut)->bool:
|
||||
|
||||
allowed={"admin","can_delete"}
|
||||
|
||||
if not any(p.permission in allowed for p in current_user.permissions):
|
||||
raise errors.forbidden_error(detail="You don't have permissions to delete users")
|
||||
|
||||
user = self.DB.delete_user(id)
|
||||
if user is False:
|
||||
raise errors.not_found_error(detail="User not found")
|
||||
return user
|
||||
|
||||
|
||||
def update_user(self, data: UserUpdate, current_user: UserOut) -> User:
|
||||
|
||||
allowed = {"admin", "can_edit"}
|
||||
is_admin = any(p.permission == "admin" for p in current_user.permissions)
|
||||
permission_names = [p.permission for p in data.permissions] if data.permissions else []
|
||||
existing_permissions = set(self.DB.get_all_permissions())
|
||||
|
||||
user_info = data.model_dump(exclude_unset=True, exclude={"permissions"})
|
||||
|
||||
|
||||
if not any(p.permission in allowed for p in current_user.permissions): #check if user has necessary permissions
|
||||
raise errors.forbidden_error(detail="You don't have permissions to update users")
|
||||
|
||||
if not set(permission_names).issubset(existing_permissions): #check if all the permissions in the data exist
|
||||
raise errors.not_found_error(detail="You have requested the unknown permission")
|
||||
|
||||
if not is_admin and "admin" in permission_names: #check for escalating permissions
|
||||
raise errors.forbidden_error(detail="You can't update user with admin permissions")
|
||||
|
||||
if self.DB.get_user_by_email(email=user_info["email"]):
|
||||
raise errors.forbidden_error(detail="You can't use this email")
|
||||
|
||||
plain_password = user_info.pop("plain_password", None)
|
||||
if plain_password is not None:
|
||||
user_info["hashed_password"] = self.hashes.plain_to_hash(plain_password) #re-hash password
|
||||
|
||||
target_id = user_info.pop("target_id", None)
|
||||
if target_id is None:
|
||||
current_user_info = self.DB.get_user_by_email(email=current_user.email)
|
||||
if current_user_info is None:
|
||||
raise errors.not_found_error(detail="Current user doesn't have an id")
|
||||
target_id = current_user_info.id
|
||||
|
||||
user_info["id"] = target_id
|
||||
|
||||
user = self.DB.update_user(data=user_info, permission_names=permission_names)
|
||||
|
||||
if user is None:
|
||||
raise errors.not_found_error(detail="User not found")
|
||||
return user
|
||||
|
||||
crud_actions=CrudActions()
|
||||
@@ -0,0 +1 @@
|
||||
#Сюда e2e тесты
|
||||
@@ -0,0 +1 @@
|
||||
#Сюда unit
|
||||
@@ -0,0 +1,56 @@
|
||||
from fastapi import APIRouter, Depends
|
||||
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
||||
from src.model.user.user_model import UserCreate, UserOut, UserUpdate
|
||||
from src.service.auth.auth import auth
|
||||
from src.service.crud_actions.crud import crud_actions
|
||||
|
||||
|
||||
router=APIRouter(prefix="/protected")
|
||||
oauth2_schema=OAuth2PasswordBearer(tokenUrl="/protected/token")
|
||||
|
||||
|
||||
@router.post("/token")
|
||||
def create_token(form_data:OAuth2PasswordRequestForm=Depends())->dict:
|
||||
|
||||
return auth.create_token(form_data_email=form_data.username, form_data_password=form_data.password)
|
||||
|
||||
|
||||
def get_current_user(token: str = Depends(oauth2_schema)) -> UserOut:
|
||||
|
||||
return UserOut.model_validate(auth.get_current_user(token))
|
||||
|
||||
|
||||
@router.get("")
|
||||
def protected(token:str=Depends(get_current_user))->dict:
|
||||
return {"message":"Hello,this is a protected route"}
|
||||
|
||||
@router.get("/logout")
|
||||
def logout(token:str=Depends(get_current_user))->dict:
|
||||
return {"message":"You have successful made a logout"}
|
||||
|
||||
@router.get("/get_user_by_email",response_model=UserOut)
|
||||
def protected_get_user_email(email:str, current_user:UserOut=Depends(get_current_user))->UserOut:
|
||||
|
||||
user = crud_actions.get_user_by_email(email=email, current_user=current_user)
|
||||
return UserOut.model_validate(user)
|
||||
|
||||
@router.get("/get_user_by_id",response_model=UserOut)
|
||||
def protected_get_user_id(id:int, current_user:UserOut=Depends(get_current_user))->UserOut:
|
||||
|
||||
user = crud_actions.get_user_by_id(id=id, current_user=current_user)
|
||||
return UserOut.model_validate(user)
|
||||
|
||||
@router.post("/create_user", response_model=UserOut)
|
||||
def create_user(data:UserCreate, current_user:UserOut=Depends(get_current_user))->UserOut:
|
||||
user=crud_actions.create_user(data=data, current_user=current_user)
|
||||
return UserOut.model_validate(user)
|
||||
|
||||
@router.post("/delete_user")
|
||||
def delete_user(id:int, current_user:UserOut=Depends(get_current_user))->bool:
|
||||
return crud_actions.delete_user(id=id, current_user=current_user)
|
||||
|
||||
|
||||
@router.patch("/update_user", response_model=UserOut)
|
||||
def update_user(data:UserUpdate, current_user:UserOut=Depends(get_current_user))->UserOut:
|
||||
user = crud_actions.update_user(data=data, current_user=current_user)
|
||||
return UserOut.model_validate(user)
|
||||
Reference in New Issue
Block a user