final
This commit is contained in:
2
run.py
2
run.py
@@ -35,7 +35,7 @@ async def arguments(args):
|
|||||||
"surname": args.user_name,
|
"surname": args.user_name,
|
||||||
"admin": True
|
"admin": True
|
||||||
}
|
}
|
||||||
users = await list_users() or []
|
users = await list_users()
|
||||||
label = any(u.admin for u in users)
|
label = any(u.admin for u in users)
|
||||||
if not label:
|
if not label:
|
||||||
await create_user(UserCreate(**admin_user))
|
await create_user(UserCreate(**admin_user))
|
||||||
|
|||||||
@@ -0,0 +1,32 @@
|
|||||||
|
"""empty message
|
||||||
|
|
||||||
|
Revision ID: 1e2bd98e74a5
|
||||||
|
Revises: 4ffe643b7d40
|
||||||
|
Create Date: 2026-03-07 00:44:25.427515
|
||||||
|
|
||||||
|
"""
|
||||||
|
from typing import Sequence, Union
|
||||||
|
|
||||||
|
from alembic import op
|
||||||
|
import sqlalchemy as sa
|
||||||
|
|
||||||
|
|
||||||
|
# revision identifiers, used by Alembic.
|
||||||
|
revision: str = '1e2bd98e74a5'
|
||||||
|
down_revision: Union[str, Sequence[str], None] = '4ffe643b7d40'
|
||||||
|
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! ###
|
||||||
|
pass
|
||||||
|
# ### end Alembic commands ###
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade() -> None:
|
||||||
|
"""Downgrade schema."""
|
||||||
|
# ### commands auto generated by Alembic - please adjust! ###
|
||||||
|
pass
|
||||||
|
# ### end Alembic commands ###
|
||||||
@@ -13,7 +13,10 @@ async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(s
|
|||||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
||||||
return user
|
return user
|
||||||
async def check_roles(user=Depends(get_current_user)):
|
async def check_roles(user=Depends(get_current_user)):
|
||||||
if user.get("admin") != True:
|
user_check = await db.list_user(user["user_id"])
|
||||||
|
if not user_check:
|
||||||
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
||||||
|
if user_check.admin != True:
|
||||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")
|
||||||
return user
|
return user
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user