329 lines
9.8 KiB
Python
329 lines
9.8 KiB
Python
# app/api/v1/admin/vendor_domains.py
|
|
"""
|
|
Admin endpoints for managing vendor custom domains.
|
|
|
|
Follows the architecture pattern:
|
|
- Endpoints only handle HTTP layer
|
|
- Business logic in service layer
|
|
- Proper exception handling
|
|
- Pydantic schemas for validation
|
|
"""
|
|
|
|
import logging
|
|
from typing import List
|
|
|
|
from fastapi import APIRouter, Depends, Path, Body, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_current_admin_user
|
|
from app.core.database import get_db
|
|
from app.services.vendor_domain_service import vendor_domain_service
|
|
from app.exceptions import VendorNotFoundException
|
|
from models.schema.vendor_domain import (
|
|
VendorDomainCreate,
|
|
VendorDomainUpdate,
|
|
VendorDomainResponse,
|
|
VendorDomainListResponse,
|
|
DomainVerificationInstructions,
|
|
DomainVerificationResponse,
|
|
DomainDeletionResponse,
|
|
)
|
|
from models.database.user import User
|
|
from models.database.vendor import Vendor
|
|
|
|
router = APIRouter(prefix="/vendors")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _get_vendor_by_id(db: Session, vendor_id: int) -> Vendor:
|
|
"""
|
|
Helper to get vendor by ID.
|
|
|
|
Args:
|
|
db: Database session
|
|
vendor_id: Vendor ID
|
|
|
|
Returns:
|
|
Vendor object
|
|
|
|
Raises:
|
|
VendorNotFoundException: If vendor not found
|
|
"""
|
|
vendor = db.query(Vendor).filter(Vendor.id == vendor_id).first()
|
|
if not vendor:
|
|
raise VendorNotFoundException(str(vendor_id), identifier_type="id")
|
|
return vendor
|
|
|
|
|
|
@router.post("/{vendor_id}/domains", response_model=VendorDomainResponse)
|
|
def add_vendor_domain(
|
|
vendor_id: int = Path(..., description="Vendor ID", gt=0),
|
|
domain_data: VendorDomainCreate = Body(...),
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
Add a custom domain to vendor (Admin only).
|
|
|
|
This endpoint:
|
|
1. Validates the domain format
|
|
2. Checks if domain is already registered
|
|
3. Generates verification token
|
|
4. Creates domain record (unverified, inactive)
|
|
5. Returns domain with verification instructions
|
|
|
|
**Domain Examples:**
|
|
- myshop.com
|
|
- shop.mybrand.com
|
|
- customstore.net
|
|
|
|
**Next Steps:**
|
|
1. Vendor adds DNS TXT record
|
|
2. Admin clicks "Verify Domain" to confirm ownership
|
|
3. Once verified, domain can be activated
|
|
|
|
**Raises:**
|
|
- 404: Vendor not found
|
|
- 409: Domain already registered
|
|
- 422: Invalid domain format or reserved subdomain
|
|
"""
|
|
domain = vendor_domain_service.add_domain(
|
|
db=db,
|
|
vendor_id=vendor_id,
|
|
domain_data=domain_data
|
|
)
|
|
|
|
return VendorDomainResponse(
|
|
id=domain.id,
|
|
vendor_id=domain.vendor_id,
|
|
domain=domain.domain,
|
|
is_primary=domain.is_primary,
|
|
is_active=domain.is_active,
|
|
is_verified=domain.is_verified,
|
|
ssl_status=domain.ssl_status,
|
|
verification_token=domain.verification_token,
|
|
verified_at=domain.verified_at,
|
|
ssl_verified_at=domain.ssl_verified_at,
|
|
created_at=domain.created_at,
|
|
updated_at=domain.updated_at,
|
|
)
|
|
|
|
|
|
@router.get("/{vendor_id}/domains", response_model=VendorDomainListResponse)
|
|
def list_vendor_domains(
|
|
vendor_id: int = Path(..., description="Vendor ID", gt=0),
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
List all domains for a vendor (Admin only).
|
|
|
|
Returns domains ordered by:
|
|
1. Primary domains first
|
|
2. Creation date (newest first)
|
|
|
|
**Raises:**
|
|
- 404: Vendor not found
|
|
"""
|
|
# Verify vendor exists
|
|
_get_vendor_by_id(db, vendor_id)
|
|
|
|
domains = vendor_domain_service.get_vendor_domains(db, vendor_id)
|
|
|
|
return VendorDomainListResponse(
|
|
domains=[
|
|
VendorDomainResponse(
|
|
id=d.id,
|
|
vendor_id=d.vendor_id,
|
|
domain=d.domain,
|
|
is_primary=d.is_primary,
|
|
is_active=d.is_active,
|
|
is_verified=d.is_verified,
|
|
ssl_status=d.ssl_status,
|
|
verification_token=d.verification_token if not d.is_verified else None,
|
|
verified_at=d.verified_at,
|
|
ssl_verified_at=d.ssl_verified_at,
|
|
created_at=d.created_at,
|
|
updated_at=d.updated_at,
|
|
)
|
|
for d in domains
|
|
],
|
|
total=len(domains)
|
|
)
|
|
|
|
|
|
@router.get("/domains/{domain_id}", response_model=VendorDomainResponse)
|
|
def get_domain_details(
|
|
domain_id: int = Path(..., description="Domain ID", gt=0),
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
Get detailed information about a specific domain (Admin only).
|
|
|
|
**Raises:**
|
|
- 404: Domain not found
|
|
"""
|
|
domain = vendor_domain_service.get_domain_by_id(db, domain_id)
|
|
|
|
return VendorDomainResponse(
|
|
id=domain.id,
|
|
vendor_id=domain.vendor_id,
|
|
domain=domain.domain,
|
|
is_primary=domain.is_primary,
|
|
is_active=domain.is_active,
|
|
is_verified=domain.is_verified,
|
|
ssl_status=domain.ssl_status,
|
|
verification_token=domain.verification_token if not domain.is_verified else None,
|
|
verified_at=domain.verified_at,
|
|
ssl_verified_at=domain.ssl_verified_at,
|
|
created_at=domain.created_at,
|
|
updated_at=domain.updated_at,
|
|
)
|
|
|
|
|
|
@router.put("/domains/{domain_id}", response_model=VendorDomainResponse)
|
|
def update_vendor_domain(
|
|
domain_id: int = Path(..., description="Domain ID", gt=0),
|
|
domain_update: VendorDomainUpdate = Body(...),
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
Update domain settings (Admin only).
|
|
|
|
**Can update:**
|
|
- `is_primary`: Set as primary domain for vendor
|
|
- `is_active`: Activate or deactivate domain
|
|
|
|
**Important:**
|
|
- Cannot activate unverified domains
|
|
- Setting a domain as primary will unset other primary domains
|
|
- Cannot modify domain name (delete and recreate instead)
|
|
|
|
**Raises:**
|
|
- 404: Domain not found
|
|
- 400: Cannot activate unverified domain
|
|
"""
|
|
domain = vendor_domain_service.update_domain(
|
|
db=db,
|
|
domain_id=domain_id,
|
|
domain_update=domain_update
|
|
)
|
|
|
|
return VendorDomainResponse(
|
|
id=domain.id,
|
|
vendor_id=domain.vendor_id,
|
|
domain=domain.domain,
|
|
is_primary=domain.is_primary,
|
|
is_active=domain.is_active,
|
|
is_verified=domain.is_verified,
|
|
ssl_status=domain.ssl_status,
|
|
verification_token=None, # Don't expose token after updates
|
|
verified_at=domain.verified_at,
|
|
ssl_verified_at=domain.ssl_verified_at,
|
|
created_at=domain.created_at,
|
|
updated_at=domain.updated_at,
|
|
)
|
|
|
|
|
|
@router.delete("/domains/{domain_id}", response_model=DomainDeletionResponse)
|
|
def delete_vendor_domain(
|
|
domain_id: int = Path(..., description="Domain ID", gt=0),
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
Delete a custom domain (Admin only).
|
|
|
|
**Warning:** This is permanent and cannot be undone.
|
|
|
|
**Raises:**
|
|
- 404: Domain not found
|
|
"""
|
|
# Get domain details before deletion
|
|
domain = vendor_domain_service.get_domain_by_id(db, domain_id)
|
|
vendor_id = domain.vendor_id
|
|
domain_name = domain.domain
|
|
|
|
# Delete domain
|
|
message = vendor_domain_service.delete_domain(db, domain_id)
|
|
|
|
return DomainDeletionResponse(
|
|
message=message,
|
|
domain=domain_name,
|
|
vendor_id=vendor_id
|
|
)
|
|
|
|
|
|
@router.post("/domains/{domain_id}/verify", response_model=DomainVerificationResponse)
|
|
def verify_domain_ownership(
|
|
domain_id: int = Path(..., description="Domain ID", gt=0),
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
Verify domain ownership via DNS TXT record (Admin only).
|
|
|
|
**Verification Process:**
|
|
1. Queries DNS for TXT record: `_wizamart-verify.{domain}`
|
|
2. Checks if verification token matches
|
|
3. If found, marks domain as verified
|
|
|
|
**Requirements:**
|
|
- Vendor must have added TXT record to their DNS
|
|
- DNS propagation may take 5-15 minutes
|
|
- Record format: `_wizamart-verify.domain.com` TXT `{token}`
|
|
|
|
**After verification:**
|
|
- Domain can be activated
|
|
- Domain will be available for routing
|
|
|
|
**Raises:**
|
|
- 404: Domain not found
|
|
- 400: Already verified, or verification failed
|
|
- 502: DNS query failed
|
|
"""
|
|
domain, message = vendor_domain_service.verify_domain(db, domain_id)
|
|
|
|
return DomainVerificationResponse(
|
|
message=message,
|
|
domain=domain.domain,
|
|
verified_at=domain.verified_at,
|
|
is_verified=domain.is_verified
|
|
)
|
|
|
|
|
|
@router.get("/domains/{domain_id}/verification-instructions", response_model=DomainVerificationInstructions)
|
|
def get_domain_verification_instructions(
|
|
domain_id: int = Path(..., description="Domain ID", gt=0),
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_user),
|
|
):
|
|
"""
|
|
Get DNS verification instructions for domain (Admin only).
|
|
|
|
Returns step-by-step instructions for:
|
|
1. Where to add DNS records
|
|
2. What TXT record to create
|
|
3. Links to common registrars
|
|
4. Verification token
|
|
|
|
**Use this endpoint to:**
|
|
- Show vendors how to verify their domain
|
|
- Get the exact TXT record values
|
|
- Access registrar links
|
|
|
|
**Raises:**
|
|
- 404: Domain not found
|
|
"""
|
|
instructions = vendor_domain_service.get_verification_instructions(db, domain_id)
|
|
|
|
return DomainVerificationInstructions(
|
|
domain=instructions["domain"],
|
|
verification_token=instructions["verification_token"],
|
|
instructions=instructions["instructions"],
|
|
txt_record=instructions["txt_record"],
|
|
common_registrars=instructions["common_registrars"]
|
|
)
|