Есть модель алхимии где лежат айдишники групп массивом
class UserPermissions(Base):
__tablename__ = "user_permissions"
can_read: Mapped[list[int]] = mapped_column(MutableList.as_mutable(ARRAY(Integer)),default=list)
can_edit: Mapped[list[int]] = mapped_column(MutableList.as_mutable(ARRAY(Integer)),default=list)
user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), primary_key=True)
user: Mapped["User"] = relationship(back_populates="permissions",)
есть на модули поделенная архитектура с routes services repostitories
в роутах беру сессию и проверяю есть ли вообще текущиий пользователь
@router.get(
"/{id}",
summary="Get Camera Group by ID",
description="Fetch details of a specific camera group by its unique ID. Optionally include associated cameras.",
response_description="A JSON object with the camera group's details",
responses={
200: {
"description": "Successfully retrieved camera group",
"content": {
"application/json": {
"example": {"data": {"id": 1, "name": "Group1", "cams": [{"id": 1, "name": "Cam1"}]}}
}
}
},
404: {"description": "Camera group not found"},
500: {"description": "Internal server error"}
}
)
async def get_camgroup(
request: Request,
id: int,
session: AsyncSession = Depends(create_session),
current_user = Depends(get_current_user),
includecams: bool = Query(False, description="Include associated cameras in the response")
) -> dict:
"""
Retrieve details of a specific camera group by ID.
Args:
request (Request): The incoming HTTP request object.
id (int): The unique identifier of the camera group.
session (AsyncSession): Database session injected via dependency.
includecams (bool): Flag to include associated cameras in the response. Defaults to False.
Returns:
dict: A dictionary with a 'data' key containing the camera group details (CamGroupOutSchema).
Raises:
HTTPException: 404 if camera group is not found, 500 if server error occurs.
"""
group = await CameraGroupService(session=session).get_by_id(id, includecams)
if group is None:
raise HTTPException(status_code=404, detail="Camera group not found")
return {"data": group}
Также есть сервис где лежит бизнесс-логика
@handle_sqlalchemy_exceptions()
async def get_all(self,includecams : bool = False):
camgrouplist = []
camgrouplist = await self.repo.get_groups()
if includecams:
camgrouplist = [CamGroupOutSchema.model_validate(camgroup) for camgroup in camgrouplist]
else:
camgrouplist = [ {"id": camgroup.id, "name": camgroup.name} for camgroup in camgrouplist]
return camgrouplist
также описан класс для получения прав пользователей
на каком этапе проверять права пользователя в сервисах или в роутере