307 lines
9.9 KiB
Python
307 lines
9.9 KiB
Python
# app/api/v1/admin/vendors.py
|
|
"""
|
|
Vendor management endpoints for admin.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query, HTTPException
|
|
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 models.schema.vendor import (
|
|
VendorListResponse,
|
|
VendorResponse,
|
|
VendorDetailResponse,
|
|
VendorCreate,
|
|
VendorCreateResponse,
|
|
VendorUpdate,
|
|
VendorTransferOwnership,
|
|
VendorTransferOwnershipResponse,
|
|
)
|
|
from models.database.user import User
|
|
|
|
router = APIRouter(prefix="/vendors")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@router.post("", response_model=VendorCreateResponse)
|
|
def create_vendor_with_owner(
|
|
vendor_data: VendorCreate,
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
Create a new vendor with owner user account (Admin only).
|
|
|
|
This endpoint:
|
|
1. Creates a new vendor record
|
|
2. Creates an owner user account with owner_email
|
|
3. Sets contact_email (defaults to owner_email if not provided)
|
|
4. Sets up default roles (Owner, Manager, Editor, Viewer)
|
|
5. Returns credentials (temporary password shown ONCE)
|
|
|
|
**Email Fields:**
|
|
- `owner_email`: Used for owner's login/authentication (stored in users.email)
|
|
- `contact_email`: Public business contact (stored in vendors.contact_email)
|
|
|
|
Returns vendor details with owner credentials.
|
|
"""
|
|
vendor, owner_user, temp_password = admin_service.create_vendor_with_owner(
|
|
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,
|
|
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 credentials
|
|
owner_email=owner_user.email,
|
|
owner_username=owner_user.username,
|
|
temporary_password=temp_password,
|
|
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: Optional[str] = Query(None, description="Search by name or vendor code"),
|
|
is_active: Optional[bool] = Query(None),
|
|
is_verified: Optional[bool] = Query(None),
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""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("/{vendor_id}", response_model=VendorDetailResponse)
|
|
def get_vendor_details(
|
|
vendor_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
Get detailed vendor information including owner details (Admin only).
|
|
|
|
Returns both:
|
|
- `contact_email` (business contact)
|
|
- `owner_email` (owner's authentication email)
|
|
"""
|
|
vendor = admin_service.get_vendor_by_id(db, vendor_id)
|
|
|
|
return VendorDetailResponse(
|
|
# Vendor fields
|
|
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 details
|
|
owner_email=vendor.owner.email,
|
|
owner_username=vendor.owner.username,
|
|
)
|
|
|
|
|
|
@router.put("/{vendor_id}", response_model=VendorDetailResponse)
|
|
def update_vendor(
|
|
vendor_id: int,
|
|
vendor_update: VendorUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
Update vendor information (Admin only).
|
|
|
|
**Can update:**
|
|
- Basic info: name, description, subdomain
|
|
- Business contact: contact_email, contact_phone, website
|
|
- Business details: business_address, tax_number
|
|
- Marketplace URLs
|
|
- Status: is_active, is_verified
|
|
|
|
**Cannot update:**
|
|
- `owner_email` (use POST /vendors/{id}/transfer-ownership)
|
|
- `vendor_code` (immutable)
|
|
- `owner_user_id` (use POST /vendors/{id}/transfer-ownership)
|
|
"""
|
|
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,
|
|
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.post("/{vendor_id}/transfer-ownership", response_model=VendorTransferOwnershipResponse)
|
|
def transfer_vendor_ownership(
|
|
vendor_id: int,
|
|
transfer_data: VendorTransferOwnership,
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
Transfer vendor ownership to another user (Admin only).
|
|
|
|
**This is a critical operation that:**
|
|
- Changes the owner_user_id
|
|
- Assigns new owner to "Owner" role
|
|
- Demotes old owner to "Manager" role (or removes them)
|
|
- Creates audit trail
|
|
|
|
⚠️ **This action is logged and should be used carefully.**
|
|
|
|
**Requires:**
|
|
- `new_owner_user_id`: ID of user who will become owner
|
|
- `confirm_transfer`: Must be true
|
|
- `transfer_reason`: Optional reason for audit trail
|
|
"""
|
|
from datetime import datetime, timezone
|
|
|
|
vendor, old_owner, new_owner = admin_service.transfer_vendor_ownership(
|
|
db, vendor_id, transfer_data
|
|
)
|
|
|
|
return VendorTransferOwnershipResponse(
|
|
message="Ownership transferred successfully",
|
|
vendor_id=vendor.id,
|
|
vendor_code=vendor.vendor_code,
|
|
vendor_name=vendor.name,
|
|
old_owner={
|
|
"id": old_owner.id,
|
|
"username": old_owner.username,
|
|
"email": old_owner.email,
|
|
},
|
|
new_owner={
|
|
"id": new_owner.id,
|
|
"username": new_owner.username,
|
|
"email": new_owner.email,
|
|
},
|
|
transferred_at=datetime.now(timezone.utc),
|
|
transfer_reason=transfer_data.transfer_reason,
|
|
)
|
|
|
|
|
|
@router.put("/{vendor_id}/verify")
|
|
def verify_vendor(
|
|
vendor_id: int,
|
|
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)}
|
|
|
|
|
|
@router.put("/{vendor_id}/status")
|
|
def toggle_vendor_status(
|
|
vendor_id: int,
|
|
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)}
|
|
|
|
|
|
@router.delete("/{vendor_id}")
|
|
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),
|
|
):
|
|
"""
|
|
Delete vendor and all associated data (Admin only).
|
|
|
|
⚠️ **WARNING: This is destructive and will delete:**
|
|
- Vendor account
|
|
- All products
|
|
- All orders
|
|
- All customers
|
|
- All team members
|
|
|
|
Requires confirmation parameter: `confirm=true`
|
|
"""
|
|
if not confirm:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Deletion requires confirmation parameter: confirm=true"
|
|
)
|
|
|
|
message = admin_service.delete_vendor(db, vendor_id)
|
|
return {"message": message}
|
|
|
|
|
|
@router.get("/stats/vendors")
|
|
def get_vendor_statistics(
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""Get vendor statistics for admin dashboard (Admin only)."""
|
|
return admin_service.get_vendor_statistics(db)
|