Спрятал управление скоупами в ручки категорий

services_backend/models/database.py

@@ -1,14 +1,20 @@
 from __future__ import annotations
 
+import logging
 from enum import Enum
 
+from fastapi_sqlalchemy import db
 from sqlalchemy import Enum as DbEnum
 from sqlalchemy import ForeignKey, Integer, String
+from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .base import Base
 
 
+logger = logging.getLogger(__name__)
+
+
 class Category(Base):
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     order: Mapped[int] = mapped_column(Integer, default=1)
@@ -17,7 +23,28 @@ class Category(Base):
     buttons: Mapped[list[Button]] = relationship(
         "Button", back_populates="category", foreign_keys="Button.category_id", order_by='Button.order'
     )
-    scopes: Mapped[list[Scope]] = relationship("Scope", back_populates="category")
+
+    _scopes: Mapped[list[Scope]] = relationship("Scope", back_populates="category", lazy='joined', cascade='delete')
+
+    @hybrid_property
+    def scopes(self) -> set[str]:
+        return set(s.name for s in self._scopes)
+
+    @scopes.inplace.setter
+    def _scopes_setter(self, value: set[str]):
+        old_scopes = self.scopes
+        new_scopes = set(value)
+
+        # Удаляем более ненужные скоупы
+        for s in self._scopes:
+            if s.name in (old_scopes - new_scopes):
+                db.session.delete(s)
+
+        # Добавляем недостающие скоупы
+        for s in new_scopes - old_scopes:
+            new_scope = Scope(category=self, name=s)
+            db.session.add(new_scope)
+            self._scopes.append(new_scope)
 
 
 class Type(str, Enum):
@@ -41,4 +68,4 @@ class Scope(Base):
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     name: Mapped[str] = mapped_column(String, nullable=True)
     category_id: Mapped[int] = mapped_column(Integer, ForeignKey("category.id"))
-    category: Mapped[Category] = relationship("Category", back_populates="scopes", foreign_keys=[category_id])
+    category: Mapped[Category] = relationship("Category", back_populates="_scopes", foreign_keys=[category_id])

services_backend/routes/base.py

@@ -7,7 +7,9 @@
 
 from .button import button
 from .category import category
-from .scope import scope
+
+
+# from .scope import scope
 
 
 settings = get_settings()
@@ -38,4 +40,3 @@
 
 app.include_router(button, prefix='/category/{category_id}/button', tags=["Button"])
 app.include_router(category, prefix='/category', tags=["Category"])
-app.include_router(scope, prefix='/category/{category_id}/scope', tags=["Scope"])

services_backend/routes/button.py

@@ -3,17 +3,51 @@
 from auth_lib.fastapi import UnionAuth
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi_sqlalchemy import db
+from pydantic import Field
 
-from ..models.database import Button, Category
-from .models.button import ButtonCreate, ButtonGet, ButtonUpdate
-from .models.category import CategoryGet
+from services_backend.models.database import Button, Category, Type
+from services_backend.schemas import Base
 
 
 logger = logging.getLogger(__name__)
 button = APIRouter()
 
+# region schemas
 
-@button.post("/", response_model=ButtonGet)
+
+class ButtonCreate(Base):
+    icon: str = Field(description='Иконка кнопки')
+    name: str = Field(description='Название кнопки')
+    link: str = Field(description='Ссылка, на которую перенаправляет кнопка')
+    type: Type = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер')
+
+
+class ButtonUpdate(Base):
+    category_id: int | None = Field(description='Айди категории')
+    icon: str | None = Field(description='Иконка кнопки')
+    name: str | None = Field(description='Название кнопки')
+    order: int | None = Field(description='Порядок, в котором отображаются кнопки')
+    link: str | None = Field(description='Ссылка, на которую перенаправляет кнопка')
+    type: Type | None = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер')
+
+
+class ButtonGet(Base):
+    id: int = Field(description='Айди кнопки')
+    icon: str | None = Field(description='Иконка кнопки')
+    name: str | None = Field(description='Название кнопки')
+    link: str | None = Field(description='Ссылка, на которую перенаправляет кнопка')
+    order: int | None = Field(description='Порядок, в котором отображаются кнопки')
+    type: Type | None = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер')
+
+
+class ButtonsGet(Base):
+    buttons: list[ButtonGet] | None
+
+
+# endregion
+
+
+@button.post("", response_model=ButtonGet)
 def create_button(
     button_inp: ButtonCreate,
     category_id: int,
@@ -39,7 +73,7 @@ def create_button(
     return button
 
 
-@button.get("/", response_model=CategoryGet)
+@button.get("", response_model=ButtonsGet)
 def get_buttons(
     category_id: int,
     user=Depends(UnionAuth(allow_none=True, auto_error=False)),

services_backend/routes/category.py

@@ -4,17 +4,56 @@
 from auth_lib.fastapi import UnionAuth
 from fastapi import APIRouter, Depends, HTTPException, Query
 from fastapi_sqlalchemy import db
-from sqlalchemy.orm import joinedload
+from pydantic import Field
 
-from ..models.database import Button, Category
-from .models.category import CategoryCreate, CategoryGet, CategoryUpdate
+from services_backend.models.database import Button, Category, Type
+from services_backend.schemas import Base
 
 
 logger = logging.getLogger(__name__)
 category = APIRouter()
 
 
-@category.post("/", response_model=CategoryGet)
+# region schemas
+
+
+class ButtonGet(Base):
+    id: int = Field(description='Айди кнопки')
+    icon: str | None = Field(description='Иконка кнопки')
+    name: str | None = Field(description='Название кнопки')
+    link: str | None = Field(description='Ссылка, на которую перенаправляет кнопка')
+    order: int | None = Field(description='Порядок, в котором отображаются кнопки')
+    type: Type | None = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер')
+
+
+class CategoryCreate(Base):
+    type: str = Field(description='Тип отображения категории')
+    name: str = Field(description='Имя категории')
+    scopes: set[str] | None = Field(description='Каким пользователям будет видна категория')
+
+
+class CategoryUpdate(Base):
+    order: int | None = Field(description='На какую позицию перенести категорию')
+    type: str | None = Field(description='Тип отображения категории')
+    name: str | None = Field(description='Имя категории')
+    scopes: set[str] | None = Field(description='Каким пользователям будет видна категория')
+
+
+class CategoryGet(Base):
+    id: int
+    order: int
+    type: str | None
+    name: str | None
+    buttons: list[ButtonGet] | None
+    scopes: list[str] | None
+
+
+# endregion
+
+# region routes
+
+
+@category.post("", response_model=CategoryGet)
 def create_category(
     category_inp: CategoryCreate,
     user=Depends(UnionAuth(['services.category.create'])),
@@ -25,15 +64,19 @@ def create_category(
     """
     logger.info(f"User {user.get('id')} triggered create_category")
     last_category = db.session.query(Category).order_by(Category.order.desc()).first()
+    scopes = category_inp.scopes
+    category_inp.scopes = None
     category = Category(**category_inp.dict(exclude_none=True))
+    if scopes is not None:
+        category.scopes = scopes
     if last_category:
         category.order = last_category.order + 1
     db.session.add(category)
-    db.session.flush()
+    db.session.commit()
     return category
 
 
-@category.get("/", response_model=list[CategoryGet], response_model_exclude_none=True)
+@category.get("", response_model=list[CategoryGet], response_model_exclude_none=True)
 def get_categories(
     info: list[Literal['buttons']] = Query([]),
     user=Depends(UnionAuth(allow_none=True, auto_error=False)),
@@ -50,9 +93,9 @@ def get_categories(
 
     user_scopes = set([scope["name"] for scope in user["session_scopes"]] if user else [])
     filtered_categories = []
-    for category in db.session.query(Category).order_by(Category.order).options(joinedload(Category.scopes)).all():
-        category_scopes = set([scope.__dict__["name"] for scope in category.scopes])
-        if (category_scopes == set()) or (user_scopes & category_scopes):
+    for category in db.session.query(Category).order_by(Category.order).all():
+        category_scopes = set(category.scopes)
+        if (category_scopes == set()) or len(category_scopes - user_scopes) != 0:
             filtered_categories.append(category)
 
     return [
@@ -78,9 +121,7 @@ def get_category(
 
     user_scopes = set([scope["name"] for scope in user["session_scopes"]] if user else [])
     category = db.session.query(Category).filter(Category.id == category_id).one_or_none()
-    if not category or (
-        category.scopes and not (user_scopes & set([scope.__dict__["name"] for scope in category.scopes]))
-    ):
+    if not category or (category.scopes and len(category.scopes - user_scopes) != 0):
         raise HTTPException(status_code=404, detail="Category does not exist")
     return {
         "id": category_id,
@@ -123,14 +164,17 @@ def update_category(
 
     Необходимые scopes: `services.category.update`
     """
+
     logger.info(f"User {user.get('id')} triggered update_category")
     category = db.session.query(Category).filter(Category.id == category_id).one_or_none()
     last_category = db.session.query(Category).order_by(Category.order.desc()).first()
 
     if not category:
         raise HTTPException(status_code=404, detail="Category does not exist")
-    if not any(category_inp.dict().values()):
-        raise HTTPException(status_code=400, detail="Empty schema")
+
+    if category_inp.scopes is not None:
+        category.scopes = category_inp.scopes
+        db.session.flush()
 
     if category_inp.order:
         if category_inp.order < 1:
@@ -148,6 +192,11 @@ def update_category(
             db.session.query(Category).filter(Category.order > category.order).update({"order": Category.order - 1})
 
     query = db.session.query(Category).filter(Category.id == category_id)
-    query.update(category_inp.dict(exclude_unset=True, exclude_none=True))
-    db.session.flush()
+    update_values = category_inp.dict(exclude_unset=True, exclude_none=True, exclude={'scopes': True})
+    if update_values:
+        query.update(update_values)
+    db.session.commit()
     return category
+
+
+# endregion