Samir Boulahtit
66c967a04e
- Add transfer-ownership endpoint to companies API - Add user search endpoint for autocomplete (/admin/users/search) - Update company detail endpoint to include owner info and vendors list - Update vendor endpoints to use company relationship for ownership - Update deps.py vendor access check to use company.owner_user_id 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
386 lines
13 KiB
Python
386 lines
13 KiB
Python
# app/api/v1/admin/vendors.py
|
|
"""
|
|
Vendor management endpoints for admin.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Body, Depends, Path, Query
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from app.api.deps import get_current_admin_api
|
|
from app.core.database import get_db
|
|
from app.exceptions import ConfirmationRequiredException, VendorNotFoundException
|
|
from app.services.admin_service import admin_service
|
|
from app.services.stats_service import stats_service
|
|
from models.database.company import Company
|
|
from models.database.user import User
|
|
from models.database.vendor import Vendor
|
|
from models.schema.stats import VendorStatsResponse
|
|
from models.schema.vendor import (
|
|
VendorCreate,
|
|
VendorCreateResponse,
|
|
VendorDetailResponse,
|
|
VendorListResponse,
|
|
VendorUpdate,
|
|
)
|
|
|
|
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)
|
|
.options(joinedload(Vendor.company).joinedload(Company.owner))
|
|
.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(
|
|
vendor_data: VendorCreate,
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_api),
|
|
):
|
|
"""
|
|
Create a new vendor (storefront/brand) under an existing company (Admin only).
|
|
|
|
This endpoint:
|
|
1. Validates that the parent company exists
|
|
2. Creates a new vendor record linked to the company
|
|
3. Sets up default roles (Owner, Manager, Editor, Viewer)
|
|
|
|
The vendor inherits owner and contact information from its parent company.
|
|
"""
|
|
vendor = admin_service.create_vendor(db=db, vendor_data=vendor_data)
|
|
|
|
return VendorCreateResponse(
|
|
# Vendor fields
|
|
id=vendor.id,
|
|
vendor_code=vendor.vendor_code,
|
|
subdomain=vendor.subdomain,
|
|
name=vendor.name,
|
|
description=vendor.description,
|
|
company_id=vendor.company_id,
|
|
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,
|
|
is_active=vendor.is_active,
|
|
is_verified=vendor.is_verified,
|
|
created_at=vendor.created_at,
|
|
updated_at=vendor.updated_at,
|
|
# Company info
|
|
company_name=vendor.company.name,
|
|
company_contact_email=vendor.company.contact_email,
|
|
company_contact_phone=vendor.company.contact_phone,
|
|
company_website=vendor.company.website,
|
|
# Owner info (from company)
|
|
owner_email=vendor.company.owner.email,
|
|
owner_username=vendor.company.owner.username,
|
|
login_url=f"http://localhost:8000/vendor/{vendor.subdomain}/login",
|
|
)
|
|
|
|
|
|
@router.get("", response_model=VendorListResponse)
|
|
def get_all_vendors_admin(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
search: str | None = Query(None, description="Search by name or vendor code"),
|
|
is_active: bool | None = Query(None),
|
|
is_verified: bool | None = Query(None),
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_api),
|
|
):
|
|
"""Get all vendors with filtering (Admin only)."""
|
|
vendors, total = admin_service.get_all_vendors(
|
|
db=db,
|
|
skip=skip,
|
|
limit=limit,
|
|
search=search,
|
|
is_active=is_active,
|
|
is_verified=is_verified,
|
|
)
|
|
return VendorListResponse(vendors=vendors, total=total, skip=skip, limit=limit)
|
|
|
|
|
|
@router.get("/stats", response_model=VendorStatsResponse)
|
|
def get_vendor_statistics_endpoint(
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_api),
|
|
):
|
|
"""Get vendor statistics for admin dashboard (Admin only)."""
|
|
stats = stats_service.get_vendor_statistics(db)
|
|
|
|
return VendorStatsResponse(
|
|
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_identifier}", response_model=VendorDetailResponse)
|
|
def get_vendor_details(
|
|
vendor_identifier: str = Path(..., description="Vendor ID or vendor_code"),
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_api),
|
|
):
|
|
"""
|
|
Get detailed vendor information including company and owner details (Admin only).
|
|
|
|
Accepts either vendor ID (integer) or vendor_code (string).
|
|
|
|
Returns vendor info with company contact details and owner info.
|
|
"""
|
|
vendor = _get_vendor_by_identifier(db, vendor_identifier)
|
|
|
|
return VendorDetailResponse(
|
|
# Vendor fields
|
|
id=vendor.id,
|
|
vendor_code=vendor.vendor_code,
|
|
subdomain=vendor.subdomain,
|
|
name=vendor.name,
|
|
description=vendor.description,
|
|
company_id=vendor.company_id,
|
|
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,
|
|
is_active=vendor.is_active,
|
|
is_verified=vendor.is_verified,
|
|
created_at=vendor.created_at,
|
|
updated_at=vendor.updated_at,
|
|
# Company info
|
|
company_name=vendor.company.name,
|
|
company_contact_email=vendor.company.contact_email,
|
|
company_contact_phone=vendor.company.contact_phone,
|
|
company_website=vendor.company.website,
|
|
# Owner details (from company)
|
|
owner_email=vendor.company.owner.email,
|
|
owner_username=vendor.company.owner.username,
|
|
)
|
|
|
|
|
|
@router.put("/{vendor_identifier}", response_model=VendorDetailResponse)
|
|
def update_vendor(
|
|
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_api),
|
|
):
|
|
"""
|
|
Update vendor information (Admin only).
|
|
|
|
Accepts either vendor ID (integer) or vendor_code (string).
|
|
|
|
**Can update:**
|
|
- Basic info: name, description, subdomain
|
|
- Marketplace URLs
|
|
- Status: is_active, is_verified
|
|
|
|
**Cannot update:**
|
|
- `vendor_code` (immutable)
|
|
- Business contact info (use company update endpoints)
|
|
"""
|
|
vendor = _get_vendor_by_identifier(db, vendor_identifier)
|
|
vendor = admin_service.update_vendor(db, vendor.id, vendor_update)
|
|
|
|
return VendorDetailResponse(
|
|
id=vendor.id,
|
|
vendor_code=vendor.vendor_code,
|
|
subdomain=vendor.subdomain,
|
|
name=vendor.name,
|
|
description=vendor.description,
|
|
company_id=vendor.company_id,
|
|
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,
|
|
is_active=vendor.is_active,
|
|
is_verified=vendor.is_verified,
|
|
created_at=vendor.created_at,
|
|
updated_at=vendor.updated_at,
|
|
# Company info
|
|
company_name=vendor.company.name,
|
|
company_contact_email=vendor.company.contact_email,
|
|
company_contact_phone=vendor.company.contact_phone,
|
|
company_website=vendor.company.website,
|
|
# Owner details (from company)
|
|
owner_email=vendor.company.owner.email,
|
|
owner_username=vendor.company.owner.username,
|
|
)
|
|
|
|
|
|
# NOTE: Ownership transfer is now at the Company level.
|
|
# Use PUT /api/v1/admin/companies/{id}/transfer-ownership instead.
|
|
# This endpoint is kept for backwards compatibility but may be removed in future versions.
|
|
|
|
|
|
@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_api),
|
|
):
|
|
"""
|
|
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,
|
|
company_id=vendor.company_id,
|
|
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,
|
|
is_active=vendor.is_active,
|
|
is_verified=vendor.is_verified,
|
|
created_at=vendor.created_at,
|
|
updated_at=vendor.updated_at,
|
|
# Company info
|
|
company_name=vendor.company.name,
|
|
company_contact_email=vendor.company.contact_email,
|
|
company_contact_phone=vendor.company.contact_phone,
|
|
company_website=vendor.company.website,
|
|
# Owner details (from company)
|
|
owner_email=vendor.company.owner.email,
|
|
owner_username=vendor.company.owner.username,
|
|
)
|
|
|
|
|
|
@router.put("/{vendor_identifier}/status", response_model=VendorDetailResponse)
|
|
def toggle_vendor_status(
|
|
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_api),
|
|
):
|
|
"""
|
|
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,
|
|
company_id=vendor.company_id,
|
|
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,
|
|
is_active=vendor.is_active,
|
|
is_verified=vendor.is_verified,
|
|
created_at=vendor.created_at,
|
|
updated_at=vendor.updated_at,
|
|
# Company info
|
|
company_name=vendor.company.name,
|
|
company_contact_email=vendor.company.contact_email,
|
|
company_contact_phone=vendor.company.contact_phone,
|
|
company_website=vendor.company.website,
|
|
# Owner details (from company)
|
|
owner_email=vendor.company.owner.email,
|
|
owner_username=vendor.company.owner.username,
|
|
)
|
|
|
|
|
|
@router.delete("/{vendor_identifier}")
|
|
def delete_vendor(
|
|
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_api),
|
|
):
|
|
"""
|
|
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
|
|
- All orders
|
|
- All customers
|
|
- All team members
|
|
|
|
Requires confirmation parameter: `confirm=true`
|
|
"""
|
|
# Raise custom exception instead of HTTPException
|
|
if not confirm:
|
|
raise ConfirmationRequiredException(
|
|
operation="delete_vendor",
|
|
message="Deletion requires confirmation parameter: confirm=true",
|
|
)
|
|
|
|
vendor = _get_vendor_by_identifier(db, vendor_identifier)
|
|
message = admin_service.delete_vendor(db, vendor.id)
|
|
return {"message": message}
|