FastAPI 12. Интеграция Starlette Admin

README.md

@@ -61,9 +61,10 @@ PostgreSQL, Poetry, Pydantic и других.
 9. [FastAPI 9. Logout и проверка авторизации](https://pressanybutton.ru/post/servis-na-fastapi/fastapi-9-logout-i-proverka-avtorizacii/)
 10. [FastAPI 10. Изменение данных пользователя](https://pressanybutton.ru/post/servis-na-fastapi/fastapi-10-izmenenie-dannyh-polzovatelya/)
 11. [FastAPI 11. Хранение и сокращение ссылок](https://pressanybutton.ru/post/servis-na-fastapi/fastapi-11-hranenie-i-sokrashenie-ssylok/)
+11. [FastAPI 12. Интеграция Starlette Admin](https://pressanybutton.ru/post/servis-na-fastapi/fastapi-12-integraciya-starlette-admin/)
 
 ## Установка
-h
+
 Для установки и запуска проекта на вашем локальном компьютере выполните следующие шаги.
 
 1. **Клонируйте репозиторий:**

pyproject.toml

@@ -6,23 +6,25 @@ authors = [
     { name = "proDream", email = "sushkoos@gmail.com" }
 ]
 readme = "README.md"
-requires-python = ">=3.12"
+requires-python = ">=3.12,<4"
 dependencies = [
-    "fastapi[standard] (>=0.115.6,<0.116.0)",
-    "uvicorn[standard] (>=0.34.0,<0.35.0)",
-    "pre-commit (>=4.0.1,<5.0.0)",
-    "sqlalchemy (>=2.0.37,<3.0.0)",
+    "fastapi[standard] (>=0.118.0,<0.120.0)",
+    "uvicorn[standard] (>=0.37.0,<0.40.0)",
+    "pre-commit (>=4.3.0,<5.0.0)",
+    "sqlalchemy (>=2.0.44,<3.0.0)",
     "asyncpg (>=0.30.0,<0.31.0)",
-    "pydantic-settings (>=2.7.1,<3.0.0)",
-    "alembic (>=1.14.0,<2.0.0)",
-    "ruff (>=0.9.0,<0.10.0)",
-    "pydantic[email] (>=2.10.5,<3.0.0)",
+    "pydantic-settings (>=2.11.0,<3.0.0)",
+    "alembic (>=1.17.0,<2.0.0)",
+    "ruff (>=0.14.0,<0.20.0)",
+    "pydantic[email] (>=2.12.1,<3.0.0)",
     "passlib (>=1.7.4,<2.0.0)",
     "bcrypt (==4.0.1)",
-    "celery (>=5.4.0,<6.0.0)",
-    "redis (>=5.2.1,<6.0.0)",
+    "celery (>=5.5.3,<6.0.0)",
+    "redis (>=6.4.0,<7.0.0)",
     "itsdangerous (>=2.2.0,<3.0.0)",
     "pyjwt (>=2.10.1,<3.0.0)",
+    "poetry-core (>=2.2.1,<3.0.0)",
+    "starlette-admin (>=0.15.1,<0.16.0)",
 ]
 
 [build-system]

src/lkeep/apps/admin/admin_auth.py

@@ -0,0 +1,80 @@
+from fastapi import HTTPException
+from pydantic import EmailStr, ValidationError
+from starlette.requests import Request
+from starlette.responses import Response
+from starlette_admin.auth import AuthProvider
+from starlette_admin.exceptions import LoginFailed
+
+from lkeep.apps.auth.handlers import AuthHandler
+from lkeep.apps.auth.managers import UserManager
+from lkeep.apps.auth.schemas import AuthUser, GetUserByID
+from lkeep.apps.auth.utils import get_token_from_cookies
+from lkeep.core.core_dependency.db_dependency import DBDependency
+from lkeep.core.core_dependency.redis_dependency import RedisDependency
+from lkeep.core.settings import settings
+
+
+class AdminAuthProvider(AuthProvider):
+    def __init__(self, handler: AuthHandler, manager: UserManager):
+        super().__init__()
+        self.handler = handler
+        self.manager = manager
+
+    async def is_authenticated(self, request: Request) -> bool:
+        try:
+            token = await get_token_from_cookies(request=request)
+            decoded_token = await self.handler.decode_access_token(token=token)
+        except HTTPException:
+            return False
+
+        user_id = str(decoded_token.get("user_id"))
+        session_id = str(decoded_token.get("session_id"))
+
+        if not await self.manager.get_access_token(user_id=user_id, session_id=session_id):
+            return False
+
+        request.state.user = {"id": user_id, "session_id": session_id}
+        return True
+
+    async def login(
+        self, email: EmailStr, password: str, remember_me: bool, request: Request, response: Response
+    ) -> Response:
+        try:
+            auth_data = AuthUser(email=email, password=password)
+        except ValidationError:
+            raise LoginFailed(msg="Invalid email or password")
+        exist_user = await self.manager.get_user_by_email_for_admin(email=auth_data.email)
+
+        if (
+            exist_user is None
+            or not exist_user.is_superuser
+            or not await self.handler.verify_password(
+                hashed_password=exist_user.hashed_password, raw_password=auth_data.password
+            )
+        ):
+            raise LoginFailed(msg="Invalid credentials")
+
+        token, session_id = await self.handler.create_access_token(user_id=exist_user.id)
+
+        await self.manager.store_access_token(token=token, user_id=exist_user.id, session_id=session_id)
+
+        response.set_cookie(key="Authorization", value=token, httponly=True, max_age=settings.access_token_expire)
+
+        return response
+
+    async def logout(self, request: Request, response: Response) -> Response:
+        response.delete_cookie("Authorization")
+
+        user = request.state.user
+
+        await self.manager.revoke_access_token(user_id=user["id"], session_id=user["session_id"])
+
+        return response
+
+    def get_admin_user(self, request: Request) -> GetUserByID:
+        return GetUserByID(**request.state.user)
+
+
+def get_admin_auth_provider() -> AdminAuthProvider:
+    manager = UserManager(db=DBDependency(), redis=RedisDependency())
+    return AdminAuthProvider(handler=AuthHandler(), manager=manager)

src/lkeep/apps/admin/admin_base.py

@@ -0,0 +1,17 @@
+from fastapi import FastAPI
+from starlette_admin.contrib.sqla import Admin
+
+from lkeep.apps.admin.admin_auth import get_admin_auth_provider
+from lkeep.apps.admin.views.link_view import LinkViewView
+from lkeep.apps.admin.views.user_view import UserView
+from lkeep.core.core_dependency.db_dependency import get_db_engine
+from lkeep.database.models import Link, User
+
+
+def setup_admin(app: FastAPI) -> None:
+    admin = Admin(engine=get_db_engine(), title="Lkeep Admin", auth_provider=get_admin_auth_provider())
+
+    admin.add_view(UserView(User))
+    admin.add_view(LinkViewView(Link))
+
+    admin.mount_to(app=app)

src/lkeep/apps/admin/views/link_view.py

@@ -0,0 +1,20 @@
+from typing import Any
+
+from starlette.requests import Request
+from starlette_admin import StringField
+from starlette_admin.contrib.sqla import ModelView
+
+from lkeep.database.models import Link
+
+
+class LinkViewView(ModelView):
+    fields = [
+        "id",
+        "full_link",
+        "short_link",
+        StringField("owner_id", label="owner_id", read_only=True),
+    ]
+
+    async def before_create(self, request: Request, data: dict[str, Any], link: Link):
+        admin_user = request.state.user
+        link.owner_id = admin_user["id"]

src/lkeep/apps/admin/views/user_view.py

@@ -0,0 +1,49 @@
+from typing import Any
+
+from email_validator import EmailSyntaxError, validate_email
+from passlib.context import CryptContext
+from starlette.requests import Request
+from starlette_admin import PasswordField
+from starlette_admin.contrib.sqla import ModelView
+from starlette_admin.exceptions import FormValidationError
+
+from lkeep.database.models import User
+
+
+class UserView(ModelView):
+    fields = [
+        "id",
+        "email",
+        PasswordField(
+            name="password",
+            label="Password",
+            required=True,
+            exclude_from_list=True,
+            exclude_from_detail=True,
+            exclude_from_edit=True,
+        ),
+        "is_active",
+        "is_verified",
+        "is_superuser",
+        "created_at",
+        "updated_at",
+    ]
+
+    exclude_fields_from_edit = ["created_at", "updated_at"]
+    exclude_fields_from_create = ["created_at", "updated_at"]
+
+    sortable_fields = ["email", "created_at"]
+    fields_default_sort = [("created_at", True)]
+    searchable_fields = ["email"]
+
+    page_size = 20
+    page_size_options = [5, 10, 25, 50, -1]
+
+    async def before_create(self, request: Request, data: dict[str, Any], user: User) -> None:
+        try:
+            validate_email(data["email"])
+        except EmailSyntaxError:
+            raise FormValidationError(errors={"email": "Invalid email"})
+
+        pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+        user.hashed_password = pwd_context.hash(data.pop("password"))

src/lkeep/apps/auth/managers.py

@@ -14,6 +14,7 @@ from sqlalchemy.exc import IntegrityError
 
 from lkeep.apps.auth.schemas import (
     CreateUser,
+    GetUserForAdmin,
     GetUserWithIDAndEmail,
     UserReturnData,
     UserVerifySchema,
@@ -98,6 +99,20 @@ class UserManager:
 
             return None
 
+    async def get_user_by_email_for_admin(self, email: str) -> GetUserForAdmin | None:
+        async with self.db.db_session() as session:
+            query = select(self.model.id, self.model.email, self.model.hashed_password, self.model.is_superuser).where(
+                self.model.email == email
+            )
+
+            result = await session.execute(query)
+            user = result.mappings().first()
+
+            if user:
+                return GetUserForAdmin(**user)
+
+            return None
+
     async def get_user_by_id(self, user_id: uuid.UUID | str) -> UserVerifySchema | None:
         """
         Возвращает информацию о пользователе по его идентификатору.

src/lkeep/apps/auth/schemas.py

@@ -83,6 +83,10 @@ class GetUserWithIDAndEmail(GetUserByID, CreateUser):
     pass
 
 
+class GetUserForAdmin(GetUserWithIDAndEmail):
+    is_superuser: bool
+
+
 class UserReturnData(GetUserByID, GetUserByEmail):
     """
     Класс для представления данных пользователя, возвращаемых из API.

src/lkeep/core/core_dependency/db_dependency.py

@@ -6,7 +6,12 @@
 https://pressanybutton.ru/category/servis-na-fastapi/
 """
 
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
 
 from lkeep.core.settings import settings
 
@@ -32,3 +37,11 @@ class DBDependency:
         :rtype: async_sessionmaker[AsyncSession]
         """
         return self._session_factory
+
+    @property
+    def db_engine(self) -> AsyncEngine:
+        return self._engine
+
+
+def get_db_engine() -> AsyncEngine:
+    return DBDependency().db_engine

src/lkeep/main.py

@@ -11,6 +11,7 @@ from fastapi import FastAPI
 from starlette.middleware.cors import CORSMiddleware
 
 from lkeep.apps import apps_router
+from lkeep.apps.admin.admin_base import setup_admin
 
 app = FastAPI()
 
@@ -24,6 +25,8 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+setup_admin(app=app)
+
 
 def start():
     """