Files
orion/app/api/v1/vendor/notifications.py
Samir Boulahtit 81bfc49f77 refactor: enforce strict architecture rules and add Pydantic response models
- Update architecture rules to be stricter (API-003 now blocks ALL exception
  raising in endpoints, not just HTTPException)
- Update get_current_vendor_api dependency to guarantee token_vendor_id presence
- Remove redundant _get_vendor_from_token helpers from all vendor API files
- Move vendor access validation to service layer methods
- Add Pydantic response models for media, notification, and payment endpoints
- Add get_active_vendor_by_code service method for public vendor lookup
- Add get_import_job_for_vendor service method with vendor validation
- Update validation script to detect exception raising patterns in endpoints

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-12-04 23:26:03 +01:00

220 lines
6.8 KiB
Python

# app/api/v1/vendor/notifications.py
"""
Vendor notification management endpoints.
Vendor Context: Uses token_vendor_id from JWT token (authenticated vendor API pattern).
The get_current_vendor_api dependency guarantees token_vendor_id is present.
"""
import logging
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from app.api.deps import get_current_vendor_api
from app.core.database import get_db
from app.services.vendor_service import vendor_service
from models.database.user import User
from models.schema.notification import (
MessageResponse,
NotificationListResponse,
NotificationSettingsResponse,
NotificationSettingsUpdate,
NotificationTemplateListResponse,
NotificationTemplateUpdate,
TestNotificationRequest,
UnreadCountResponse,
)
router = APIRouter(prefix="/notifications")
logger = logging.getLogger(__name__)
@router.get("", response_model=NotificationListResponse)
def get_notifications(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
unread_only: bool | None = Query(False),
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Get vendor notifications.
TODO: Implement in Slice 5
- Get all notifications for vendor
- Filter by read/unread status
- Support pagination
- Return notification details
"""
vendor = vendor_service.get_vendor_by_id(db, current_user.token_vendor_id) # noqa: F841
return NotificationListResponse(
notifications=[],
total=0,
unread_count=0,
message="Notifications coming in Slice 5",
)
@router.get("/unread-count", response_model=UnreadCountResponse)
def get_unread_count(
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Get count of unread notifications.
TODO: Implement in Slice 5
- Count unread notifications for vendor
- Used for notification badge
"""
vendor = vendor_service.get_vendor_by_id(db, current_user.token_vendor_id) # noqa: F841
return UnreadCountResponse(unread_count=0, message="Unread count coming in Slice 5")
@router.put("/{notification_id}/read", response_model=MessageResponse)
def mark_as_read(
notification_id: int,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Mark notification as read.
TODO: Implement in Slice 5
- Mark single notification as read
- Update read timestamp
"""
vendor = vendor_service.get_vendor_by_id(db, current_user.token_vendor_id) # noqa: F841
return MessageResponse(message="Mark as read coming in Slice 5")
@router.put("/mark-all-read", response_model=MessageResponse)
def mark_all_as_read(
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Mark all notifications as read.
TODO: Implement in Slice 5
- Mark all vendor notifications as read
- Update timestamps
"""
vendor = vendor_service.get_vendor_by_id(db, current_user.token_vendor_id) # noqa: F841
return MessageResponse(message="Mark all as read coming in Slice 5")
@router.delete("/{notification_id}", response_model=MessageResponse)
def delete_notification(
notification_id: int,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Delete notification.
TODO: Implement in Slice 5
- Delete single notification
- Verify notification belongs to vendor
"""
vendor = vendor_service.get_vendor_by_id(db, current_user.token_vendor_id) # noqa: F841
return MessageResponse(message="Notification deletion coming in Slice 5")
@router.get("/settings", response_model=NotificationSettingsResponse)
def get_notification_settings(
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Get notification preferences.
TODO: Implement in Slice 5
- Get email notification settings
- Get in-app notification settings
- Get notification types enabled/disabled
"""
vendor = vendor_service.get_vendor_by_id(db, current_user.token_vendor_id) # noqa: F841
return NotificationSettingsResponse(
email_notifications=True,
in_app_notifications=True,
notification_types={},
message="Notification settings coming in Slice 5",
)
@router.put("/settings", response_model=MessageResponse)
def update_notification_settings(
settings: NotificationSettingsUpdate,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Update notification preferences.
TODO: Implement in Slice 5
- Update email notification settings
- Update in-app notification settings
- Enable/disable specific notification types
"""
vendor = vendor_service.get_vendor_by_id(db, current_user.token_vendor_id) # noqa: F841
return MessageResponse(message="Notification settings update coming in Slice 5")
@router.get("/templates", response_model=NotificationTemplateListResponse)
def get_notification_templates(
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Get notification email templates.
TODO: Implement in Slice 5
- Get all notification templates
- Include: order confirmation, shipping notification, etc.
- Return template details
"""
vendor = vendor_service.get_vendor_by_id(db, current_user.token_vendor_id) # noqa: F841
return NotificationTemplateListResponse(
templates=[], message="Notification templates coming in Slice 5"
)
@router.put("/templates/{template_id}", response_model=MessageResponse)
def update_notification_template(
template_id: int,
template_data: NotificationTemplateUpdate,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Update notification email template.
TODO: Implement in Slice 5
- Update template subject
- Update template body (HTML/text)
- Validate template variables
- Preview template
"""
vendor = vendor_service.get_vendor_by_id(db, current_user.token_vendor_id) # noqa: F841
return MessageResponse(message="Template update coming in Slice 5")
@router.post("/test", response_model=MessageResponse)
def send_test_notification(
notification_data: TestNotificationRequest,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Send test notification.
TODO: Implement in Slice 5
- Send test email notification
- Use specified template
- Send to current user's email
"""
vendor = vendor_service.get_vendor_by_id(db, current_user.token_vendor_id) # noqa: F841
return MessageResponse(message="Test notification coming in Slice 5")