fix vendor edit page in admin panel

This commit is contained in:
2025-10-25 11:06:06 +02:00
parent 1a43a4250c
commit 7bb69a9a96

View File

@@ -6,12 +6,14 @@ Vendor management endpoints for admin.
import logging
from typing import Optional
from fastapi import APIRouter, Depends, Query, HTTPException
from fastapi import APIRouter, Depends, Query, Path, HTTPException, Body
from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_user
from app.core.database import get_db
from app.services.admin_service import admin_service
from app.services.stats_service import stats_service
from app.exceptions import VendorNotFoundException
from models.schema.stats import VendorStatsResponse
from models.schema.vendor import (
VendorListResponse,
@@ -24,11 +26,50 @@ from models.schema.vendor import (
VendorTransferOwnershipResponse,
)
from models.database.user import User
from models.database.vendor import Vendor
from sqlalchemy import func
router = APIRouter(prefix="/vendors")
logger = logging.getLogger(__name__)
def _get_vendor_by_identifier(db: Session, identifier: str) -> Vendor:
"""
Helper to get vendor by ID or vendor_code.
Follows the pattern from admin_service._get_vendor_by_id_or_raise.
Args:
db: Database session
identifier: Either vendor ID (int as string) or vendor_code (string)
Returns:
Vendor object
Raises:
VendorNotFoundException: If vendor not found
"""
# Try as integer ID first
try:
vendor_id = int(identifier)
return admin_service.get_vendor_by_id(db, vendor_id)
except (ValueError, TypeError):
# Not an integer, treat as vendor_code
pass
except VendorNotFoundException:
# ID not found, try as vendor_code
pass
# Try as vendor_code (case-insensitive)
vendor = db.query(Vendor).filter(
func.upper(Vendor.vendor_code) == identifier.upper()
).first()
if not vendor:
raise VendorNotFoundException(identifier, identifier_type="code")
return vendor
@router.post("", response_model=VendorCreateResponse)
def create_vendor_with_owner(
vendor_data: VendorCreate,
@@ -108,34 +149,37 @@ def get_all_vendors_admin(
@router.get("/stats", response_model=VendorStatsResponse)
def get_vendor_statistics(
def get_vendor_statistics_endpoint(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""Get vendor statistics for admin dashboard (Admin only)."""
stats = get_vendor_statistics(db)
stats = stats_service.get_vendor_statistics(db)
return VendorStatsResponse(
total=stats["total_vendors"],
verified=stats["verified_vendors"],
pending=stats["total_vendors"] - stats["verified_vendors"],
inactive=stats["inactive_vendors"],
total=stats.get("total_vendors", 0),
verified=stats.get("verified_vendors", 0),
pending=stats.get("pending_vendors", 0),
inactive=stats.get("inactive_vendors", 0),
)
@router.get("/{vendor_id}", response_model=VendorDetailResponse)
@router.get("/{vendor_identifier}", response_model=VendorDetailResponse)
def get_vendor_details(
vendor_id: int,
vendor_identifier: str = Path(..., description="Vendor ID or vendor_code"),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""
Get detailed vendor information including owner details (Admin only).
Accepts either vendor ID (integer) or vendor_code (string).
Returns both:
- `contact_email` (business contact)
- `owner_email` (owner's authentication email)
"""
vendor = admin_service.get_vendor_by_id(db, vendor_id)
vendor = _get_vendor_by_identifier(db, vendor_identifier)
return VendorDetailResponse(
# Vendor fields
@@ -164,16 +208,18 @@ def get_vendor_details(
)
@router.put("/{vendor_id}", response_model=VendorDetailResponse)
@router.put("/{vendor_identifier}", response_model=VendorDetailResponse)
def update_vendor(
vendor_id: int,
vendor_update: VendorUpdate,
vendor_identifier: str = Path(..., description="Vendor ID or vendor_code"),
vendor_update: VendorUpdate = Body(...),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""
Update vendor information (Admin only).
Accepts either vendor ID (integer) or vendor_code (string).
**Can update:**
- Basic info: name, description, subdomain
- Business contact: contact_email, contact_phone, website
@@ -186,7 +232,8 @@ def update_vendor(
- `vendor_code` (immutable)
- `owner_user_id` (use POST /vendors/{id}/transfer-ownership)
"""
vendor = admin_service.update_vendor(db, vendor_id, vendor_update)
vendor = _get_vendor_by_identifier(db, vendor_identifier)
vendor = admin_service.update_vendor(db, vendor.id, vendor_update)
return VendorDetailResponse(
id=vendor.id,
@@ -213,16 +260,18 @@ def update_vendor(
)
@router.post("/{vendor_id}/transfer-ownership", response_model=VendorTransferOwnershipResponse)
@router.post("/{vendor_identifier}/transfer-ownership", response_model=VendorTransferOwnershipResponse)
def transfer_vendor_ownership(
vendor_id: int,
transfer_data: VendorTransferOwnership,
vendor_identifier: str = Path(..., description="Vendor ID or vendor_code"),
transfer_data: VendorTransferOwnership = Body(...),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""
Transfer vendor ownership to another user (Admin only).
Accepts either vendor ID (integer) or vendor_code (string).
**This is a critical operation that:**
- Changes the owner_user_id
- Assigns new owner to "Owner" role
@@ -238,8 +287,9 @@ def transfer_vendor_ownership(
"""
from datetime import datetime, timezone
vendor = _get_vendor_by_identifier(db, vendor_identifier)
vendor, old_owner, new_owner = admin_service.transfer_vendor_ownership(
db, vendor_id, transfer_data
db, vendor.id, transfer_data
)
return VendorTransferOwnershipResponse(
@@ -262,38 +312,122 @@ def transfer_vendor_ownership(
)
@router.put("/{vendor_id}/verify")
def verify_vendor(
vendor_id: int,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
@router.put("/{vendor_identifier}/verification", response_model=VendorDetailResponse)
def toggle_vendor_verification(
vendor_identifier: str = Path(..., description="Vendor ID or vendor_code"),
verification_data: dict = Body(..., example={"is_verified": True}),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""Verify/unverify vendor (Admin only)."""
vendor, message = admin_service.verify_vendor(db, vendor_id)
return {"message": message, "vendor": VendorResponse.model_validate(vendor)}
"""
Toggle vendor verification status (Admin only).
Accepts either vendor ID (integer) or vendor_code (string).
Request body: { "is_verified": true/false }
"""
vendor = _get_vendor_by_identifier(db, vendor_identifier)
# Use admin_service method if available, otherwise update directly
if "is_verified" in verification_data:
try:
vendor, message = admin_service.verify_vendor(db, vendor.id)
logger.info(f"Vendor verification toggled: {message}")
except AttributeError:
# If verify_vendor method doesn't exist, update directly
vendor.is_verified = verification_data["is_verified"]
db.commit()
db.refresh(vendor)
return VendorDetailResponse(
id=vendor.id,
vendor_code=vendor.vendor_code,
subdomain=vendor.subdomain,
name=vendor.name,
description=vendor.description,
owner_user_id=vendor.owner_user_id,
contact_email=vendor.contact_email,
contact_phone=vendor.contact_phone,
website=vendor.website,
business_address=vendor.business_address,
tax_number=vendor.tax_number,
letzshop_csv_url_fr=vendor.letzshop_csv_url_fr,
letzshop_csv_url_en=vendor.letzshop_csv_url_en,
letzshop_csv_url_de=vendor.letzshop_csv_url_de,
theme_config=vendor.theme_config or {},
is_active=vendor.is_active,
is_verified=vendor.is_verified,
created_at=vendor.created_at,
updated_at=vendor.updated_at,
owner_email=vendor.owner.email,
owner_username=vendor.owner.username,
)
@router.put("/{vendor_id}/status")
@router.put("/{vendor_identifier}/status", response_model=VendorDetailResponse)
def toggle_vendor_status(
vendor_id: int,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
vendor_identifier: str = Path(..., description="Vendor ID or vendor_code"),
status_data: dict = Body(..., example={"is_active": True}),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""Toggle vendor active status (Admin only)."""
vendor, message = admin_service.toggle_vendor_status(db, vendor_id)
return {"message": message, "vendor": VendorResponse.model_validate(vendor)}
"""
Toggle vendor active status (Admin only).
Accepts either vendor ID (integer) or vendor_code (string).
Request body: { "is_active": true/false }
"""
vendor = _get_vendor_by_identifier(db, vendor_identifier)
# Use admin_service method if available, otherwise update directly
if "is_active" in status_data:
try:
vendor, message = admin_service.toggle_vendor_status(db, vendor.id)
logger.info(f"Vendor status toggled: {message}")
except AttributeError:
# If toggle_vendor_status method doesn't exist, update directly
vendor.is_active = status_data["is_active"]
db.commit()
db.refresh(vendor)
return VendorDetailResponse(
id=vendor.id,
vendor_code=vendor.vendor_code,
subdomain=vendor.subdomain,
name=vendor.name,
description=vendor.description,
owner_user_id=vendor.owner_user_id,
contact_email=vendor.contact_email,
contact_phone=vendor.contact_phone,
website=vendor.website,
business_address=vendor.business_address,
tax_number=vendor.tax_number,
letzshop_csv_url_fr=vendor.letzshop_csv_url_fr,
letzshop_csv_url_en=vendor.letzshop_csv_url_en,
letzshop_csv_url_de=vendor.letzshop_csv_url_de,
theme_config=vendor.theme_config or {},
is_active=vendor.is_active,
is_verified=vendor.is_verified,
created_at=vendor.created_at,
updated_at=vendor.updated_at,
owner_email=vendor.owner.email,
owner_username=vendor.owner.username,
)
@router.delete("/{vendor_id}")
@router.delete("/{vendor_identifier}")
def delete_vendor(
vendor_id: int,
confirm: bool = Query(False, description="Must be true to confirm deletion"),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
vendor_identifier: str = Path(..., description="Vendor ID or vendor_code"),
confirm: bool = Query(False, description="Must be true to confirm deletion"),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""
Delete vendor and all associated data (Admin only).
Accepts either vendor ID (integer) or vendor_code (string).
⚠️ **WARNING: This is destructive and will delete:**
- Vendor account
- All products
@@ -309,6 +443,6 @@ def delete_vendor(
detail="Deletion requires confirmation parameter: confirm=true"
)
message = admin_service.delete_vendor(db, vendor_id)
return {"message": message}
vendor = _get_vendor_by_identifier(db, vendor_identifier)
message = admin_service.delete_vendor(db, vendor.id)
return {"message": message}