Replies: 2 comments 1 reply
-
There is an example in the Guards docs that demonstrates a method of restricting a particular route to admin users: from enum import Enum
from pydantic import BaseModel, UUID4
from litestar import post
from litestar.connection import ASGIConnection
from litestar.exceptions import NotAuthorizedException
from litestar.handlers.base import BaseRouteHandler
class UserRole(str, Enum):
CONSUMER = "consumer"
ADMIN = "admin"
class User(BaseModel):
id: UUID4
role: UserRole
@property
def is_admin(self) -> bool:
"""Determines whether the user is an admin user"""
return self.role == UserRole.ADMIN
def admin_user_guard(connection: ASGIConnection, _: BaseRouteHandler) -> None:
if not connection.user.is_admin:
raise NotAuthorizedException()
@post(path="/user", guards=[admin_user_guard])
def create_user(data: User) -> User: ... |
Beta Was this translation helpful? Give feedback.
-
@varuzam I haven't used SQLAdmin, but Example codefrom litestar import Litestar, Request
from sqladmin import ModelView
from sqladmin.authentication import AuthenticationBackend
from sqladmin_litestar_plugin import SQLAdminPlugin
from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import declarative_base
engine = create_async_engine("sqlite+aiosqlite:///example.db")
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String)
# eg a column to check if the user is a superuser or
# you can check however you want
is_superuser = Column(Boolean, default=False)
class AdminAuth(AuthenticationBackend):
async def login(self, request: Request) -> bool:
form = await request.form()
username, password = form["username"], form["password"]
# This is just an example. In reality you should
# check the user in the database and if the user is
# admin you can login otherwise not
if username == "admin" and password == "admin123":
# Validate username/password credentials
# And update session
request.session.update({"token": "..."})
return True
return False
async def logout(self, request: Request) -> bool:
# Usually you'd want to just clear the session
request.session.clear()
return True
async def authenticate(self, request: Request) -> bool:
token = request.session.get("token")
if not token:
return False
# Check the token in depth
return True
class UserAdmin(ModelView, model=User):
column_list = [User.id, User.name, User.is_superuser]
async def on_startup() -> None:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Create tables
admin = SQLAdminPlugin(
views=[UserAdmin],
engine=engine,
authentication_backend=AdminAuth(secret_key="some_secret_key_from_enviroment"),
)
app = Litestar(plugins=[admin], on_startup=[on_startup]) If you start uvicorn and go to |
Beta Was this translation helpful? Give feedback.
-
hello
I need to restrict the route /admin backed by a plugin sqladmin-litestar-plugin from all users except user role Admin. How to do this?
Beta Was this translation helpful? Give feedback.
All reactions