Jeroen SchweitzerandClaude Opus 4.5
381d43b60b
Build and Push / build (release) Successful in 1m28s
- Dashboard domain with Quick Links CRUD + reorder endpoints - Dashboard widgets management endpoints - Database migrations for quick_links and dashboard_widgets tables - Static file controller for Organizr widgets - Default local user when OIDC is disabled - Domain-based architecture refactor (src/domains/, src/shared/) - Test suite updated for new structure (285 tests passing) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
320 lines
9.9 KiB
Python
320 lines
9.9 KiB
Python
"""
|
|
Dashboard Domain Service
|
|
|
|
Business logic for dashboard operations.
|
|
"""
|
|
from typing import Optional, List
|
|
from sqlalchemy import select, update, delete
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.shared.logging import get_logger
|
|
from src.domains.dashboard.models import QuickLink, DashboardWidget
|
|
from src.domains.dashboard.schemas import (
|
|
QuickLinkCreate,
|
|
QuickLinkUpdate,
|
|
QuickLinkResponse,
|
|
DashboardWidgetCreate,
|
|
DashboardWidgetUpdate,
|
|
DashboardWidgetResponse,
|
|
)
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class DashboardService:
|
|
"""Service for dashboard operations"""
|
|
|
|
# =========================================================================
|
|
# Quick Links
|
|
# =========================================================================
|
|
|
|
async def get_quick_links(
|
|
self,
|
|
session: AsyncSession,
|
|
user_id: Optional[str] = None,
|
|
include_global: bool = True,
|
|
category: Optional[str] = None,
|
|
visible_only: bool = True,
|
|
) -> List[QuickLink]:
|
|
"""
|
|
Get quick links for a user
|
|
|
|
Args:
|
|
session: Database session
|
|
user_id: User ID to filter by (None for global only)
|
|
include_global: Whether to include global links (user_id=None)
|
|
category: Optional category filter
|
|
visible_only: Only return visible links
|
|
"""
|
|
conditions = []
|
|
|
|
if user_id:
|
|
if include_global:
|
|
from sqlalchemy import or_
|
|
conditions.append(or_(QuickLink.user_id == user_id, QuickLink.user_id.is_(None)))
|
|
else:
|
|
conditions.append(QuickLink.user_id == user_id)
|
|
else:
|
|
conditions.append(QuickLink.user_id.is_(None))
|
|
|
|
if category:
|
|
conditions.append(QuickLink.category == category)
|
|
|
|
if visible_only:
|
|
conditions.append(QuickLink.is_visible == True)
|
|
|
|
stmt = select(QuickLink).where(*conditions).order_by(QuickLink.position, QuickLink.id)
|
|
result = await session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_quick_link(
|
|
self,
|
|
session: AsyncSession,
|
|
link_id: int,
|
|
user_id: Optional[str] = None,
|
|
) -> Optional[QuickLink]:
|
|
"""Get a specific quick link by ID"""
|
|
conditions = [QuickLink.id == link_id]
|
|
|
|
if user_id:
|
|
from sqlalchemy import or_
|
|
conditions.append(or_(QuickLink.user_id == user_id, QuickLink.user_id.is_(None)))
|
|
|
|
stmt = select(QuickLink).where(*conditions)
|
|
result = await session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def create_quick_link(
|
|
self,
|
|
session: AsyncSession,
|
|
data: QuickLinkCreate,
|
|
user_id: Optional[str] = None,
|
|
) -> QuickLink:
|
|
"""Create a new quick link"""
|
|
# Get max position for this user
|
|
stmt = select(QuickLink.position).where(
|
|
QuickLink.user_id == user_id if user_id else QuickLink.user_id.is_(None)
|
|
).order_by(QuickLink.position.desc()).limit(1)
|
|
result = await session.execute(stmt)
|
|
max_pos = result.scalar_one_or_none() or -1
|
|
|
|
link = QuickLink(
|
|
title=data.title,
|
|
url=data.url,
|
|
icon=data.icon,
|
|
description=data.description,
|
|
category=data.category,
|
|
position=data.position if data.position > 0 else max_pos + 1,
|
|
is_visible=data.is_visible,
|
|
color=data.color,
|
|
background_color=data.background_color,
|
|
user_id=user_id,
|
|
)
|
|
session.add(link)
|
|
await session.commit()
|
|
await session.refresh(link)
|
|
|
|
logger.info(f"Created quick link: {link.title} (id={link.id}, user={user_id})")
|
|
return link
|
|
|
|
async def update_quick_link(
|
|
self,
|
|
session: AsyncSession,
|
|
link_id: int,
|
|
data: QuickLinkUpdate,
|
|
user_id: Optional[str] = None,
|
|
) -> Optional[QuickLink]:
|
|
"""Update a quick link"""
|
|
link = await self.get_quick_link(session, link_id, user_id)
|
|
if not link:
|
|
return None
|
|
|
|
# Only allow updating own links or global links for admins
|
|
if link.user_id and link.user_id != user_id:
|
|
return None
|
|
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(link, field, value)
|
|
|
|
await session.commit()
|
|
await session.refresh(link)
|
|
|
|
logger.info(f"Updated quick link: {link.title} (id={link.id})")
|
|
return link
|
|
|
|
async def delete_quick_link(
|
|
self,
|
|
session: AsyncSession,
|
|
link_id: int,
|
|
user_id: Optional[str] = None,
|
|
) -> bool:
|
|
"""Delete a quick link"""
|
|
link = await self.get_quick_link(session, link_id, user_id)
|
|
if not link:
|
|
return False
|
|
|
|
# Only allow deleting own links
|
|
if link.user_id and link.user_id != user_id:
|
|
return False
|
|
|
|
await session.delete(link)
|
|
await session.commit()
|
|
|
|
logger.info(f"Deleted quick link: id={link_id}")
|
|
return True
|
|
|
|
async def reorder_quick_links(
|
|
self,
|
|
session: AsyncSession,
|
|
link_ids: List[int],
|
|
user_id: Optional[str] = None,
|
|
) -> int:
|
|
"""Reorder quick links by updating positions"""
|
|
reordered = 0
|
|
|
|
for position, link_id in enumerate(link_ids):
|
|
conditions = [QuickLink.id == link_id]
|
|
if user_id:
|
|
from sqlalchemy import or_
|
|
conditions.append(or_(QuickLink.user_id == user_id, QuickLink.user_id.is_(None)))
|
|
|
|
stmt = update(QuickLink).where(*conditions).values(position=position)
|
|
result = await session.execute(stmt)
|
|
reordered += result.rowcount
|
|
|
|
await session.commit()
|
|
|
|
logger.info(f"Reordered {reordered} quick links for user {user_id}")
|
|
return reordered
|
|
|
|
# =========================================================================
|
|
# Dashboard Widgets
|
|
# =========================================================================
|
|
|
|
async def get_widgets(
|
|
self,
|
|
session: AsyncSession,
|
|
user_id: Optional[str] = None,
|
|
include_defaults: bool = True,
|
|
visible_only: bool = True,
|
|
) -> List[DashboardWidget]:
|
|
"""Get dashboard widgets for a user"""
|
|
conditions = []
|
|
|
|
if user_id:
|
|
if include_defaults:
|
|
from sqlalchemy import or_
|
|
conditions.append(or_(DashboardWidget.user_id == user_id, DashboardWidget.user_id.is_(None)))
|
|
else:
|
|
conditions.append(DashboardWidget.user_id == user_id)
|
|
else:
|
|
conditions.append(DashboardWidget.user_id.is_(None))
|
|
|
|
if visible_only:
|
|
conditions.append(DashboardWidget.is_visible == True)
|
|
|
|
stmt = select(DashboardWidget).where(*conditions).order_by(
|
|
DashboardWidget.position_y, DashboardWidget.position_x
|
|
)
|
|
result = await session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_widget(
|
|
self,
|
|
session: AsyncSession,
|
|
widget_id: int,
|
|
user_id: Optional[str] = None,
|
|
) -> Optional[DashboardWidget]:
|
|
"""Get a specific widget by ID"""
|
|
conditions = [DashboardWidget.id == widget_id]
|
|
|
|
if user_id:
|
|
from sqlalchemy import or_
|
|
conditions.append(or_(DashboardWidget.user_id == user_id, DashboardWidget.user_id.is_(None)))
|
|
|
|
stmt = select(DashboardWidget).where(*conditions)
|
|
result = await session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def create_widget(
|
|
self,
|
|
session: AsyncSession,
|
|
data: DashboardWidgetCreate,
|
|
user_id: Optional[str] = None,
|
|
) -> DashboardWidget:
|
|
"""Create a new dashboard widget"""
|
|
widget = DashboardWidget(
|
|
widget_type=data.widget_type,
|
|
position_x=data.position_x,
|
|
position_y=data.position_y,
|
|
width=data.width,
|
|
height=data.height,
|
|
config=data.config,
|
|
is_visible=data.is_visible,
|
|
user_id=user_id,
|
|
)
|
|
session.add(widget)
|
|
await session.commit()
|
|
await session.refresh(widget)
|
|
|
|
logger.info(f"Created widget: {widget.widget_type} (id={widget.id}, user={user_id})")
|
|
return widget
|
|
|
|
async def update_widget(
|
|
self,
|
|
session: AsyncSession,
|
|
widget_id: int,
|
|
data: DashboardWidgetUpdate,
|
|
user_id: Optional[str] = None,
|
|
) -> Optional[DashboardWidget]:
|
|
"""Update a dashboard widget"""
|
|
widget = await self.get_widget(session, widget_id, user_id)
|
|
if not widget:
|
|
return None
|
|
|
|
if widget.user_id and widget.user_id != user_id:
|
|
return None
|
|
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(widget, field, value)
|
|
|
|
await session.commit()
|
|
await session.refresh(widget)
|
|
|
|
logger.info(f"Updated widget: id={widget.id}")
|
|
return widget
|
|
|
|
async def delete_widget(
|
|
self,
|
|
session: AsyncSession,
|
|
widget_id: int,
|
|
user_id: Optional[str] = None,
|
|
) -> bool:
|
|
"""Delete a dashboard widget"""
|
|
widget = await self.get_widget(session, widget_id, user_id)
|
|
if not widget:
|
|
return False
|
|
|
|
if widget.user_id and widget.user_id != user_id:
|
|
return False
|
|
|
|
await session.delete(widget)
|
|
await session.commit()
|
|
|
|
logger.info(f"Deleted widget: id={widget_id}")
|
|
return True
|
|
|
|
|
|
# Singleton instance
|
|
_dashboard_service: Optional[DashboardService] = None
|
|
|
|
|
|
def get_dashboard_service() -> DashboardService:
|
|
"""Get singleton dashboard service instance"""
|
|
global _dashboard_service
|
|
if _dashboard_service is None:
|
|
_dashboard_service = DashboardService()
|
|
return _dashboard_service
|