sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: migrate products and vendor_products to module auto-discovery

Browse Source 
- Move admin/products.py to marketplace module as admin_products.py
  (marketplace product catalog browsing)
- Move admin/vendor_products.py to catalog module as admin.py
  (vendor catalog management)
- Move vendor/products.py to catalog module as vendor.py
  (vendor's own product catalog)
- Update marketplace admin router to include products routes
- Update catalog module routes/api/__init__.py with lazy imports
- Remove legacy imports from admin and vendor API init files

All product routes now auto-discovered via module system.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-01-31 12:49:11 +01:00
parent db56b34894
commit 6f278131a3
7 changed files with 92 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
app/api/v1/admin/__init__.py
View File

@@ -29,7 +29,8 @@ Self-contained modules (auto-discovered from app/modules/{module}/routes/api/adm
- billing: Subscription tiers, vendor billing, invoices
- inventory: Stock management, inventory tracking
- orders: Order management, fulfillment, exceptions
- marketplace: Letzshop integration, product sync
- marketplace: Letzshop integration, product sync, marketplace products
- catalog: Vendor product catalog management
- cms: Content pages management
- customers: Customer management
"""
@@ -58,13 +59,11 @@ from . import (
notifications,
platform_health,
platforms,
products,
settings,
subscriptions, # Legacy - will be replaced by billing module router
tests,
users,
vendor_domains,
vendor_products,
vendor_themes,
vendors,
)
@@ -129,17 +128,6 @@ router.include_router(admin_users.router, tags=["admin-admin-users"])
router.include_router(dashboard.router, tags=["admin-dashboard"])
# ============================================================================
# Vendor Operations (Product Catalog)
# ============================================================================
# Include marketplace product catalog management endpoints
router.include_router(products.router, tags=["admin-marketplace-products"])
# Include vendor product catalog management endpoints
router.include_router(vendor_products.router, tags=["admin-vendor-products"])
# ============================================================================
# Platform Administration
# ============================================================================

260
app/api/v1/admin/products.py
View File

@@ -1,260 +0,0 @@
# app/api/v1/admin/products.py
"""
Admin product catalog endpoints.
Provides platform-wide product search and management capabilities:
- Browse all marketplace products across vendors
- Search by title, GTIN, SKU, brand
- Filter by marketplace, vendor, availability, product type
- View product details and translations
- Copy products to vendor catalogs
"""
import logging
from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_api
from app.core.database import get_db
from app.services.marketplace_product_service import marketplace_product_service
from models.schema.auth import UserContext
router = APIRouter(prefix="/products")
logger = logging.getLogger(__name__)
# ============================================================================
# Pydantic Models
# ============================================================================
class AdminProductListItem(BaseModel):
"""Product item for admin list view."""
id: int
marketplace_product_id: str
title: str | None = None
brand: str | None = None
gtin: str | None = None
sku: str | None = None
marketplace: str | None = None
vendor_name: str | None = None
price_numeric: float | None = None
currency: str | None = None
availability: str | None = None
image_link: str | None = None
is_active: bool | None = None
is_digital: bool | None = None
product_type_enum: str | None = None
created_at: str | None = None
updated_at: str | None = None
class Config:
from_attributes = True
class AdminProductListResponse(BaseModel):
"""Paginated product list response for admin."""
products: list[AdminProductListItem]
total: int
skip: int
limit: int
class AdminProductStats(BaseModel):
"""Product statistics for admin dashboard."""
total: int
active: int
inactive: int
digital: int
physical: int
by_marketplace: dict[str, int]
class MarketplacesResponse(BaseModel):
"""Response for marketplaces list."""
marketplaces: list[str]
class VendorsResponse(BaseModel):
"""Response for vendors list."""
vendors: list[str]
class CopyToVendorRequest(BaseModel):
"""Request body for copying products to vendor catalog."""
marketplace_product_ids: list[int]
vendor_id: int
skip_existing: bool = True
class CopyToVendorResponse(BaseModel):
"""Response from copy to vendor operation."""
copied: int
skipped: int
failed: int
auto_matched: int = 0
limit_reached: bool = False
details: list[dict] | None = None
class AdminProductDetail(BaseModel):
"""Detailed product information for admin view."""
id: int
marketplace_product_id: str | None = None
gtin: str | None = None
mpn: str | None = None
sku: str | None = None
brand: str | None = None
marketplace: str | None = None
vendor_name: str | None = None
source_url: str | None = None
price: str | None = None
price_numeric: float | None = None
sale_price: str | None = None
sale_price_numeric: float | None = None
currency: str | None = None
availability: str | None = None
condition: str | None = None
image_link: str | None = None
additional_images: list | None = None
is_active: bool | None = None
is_digital: bool | None = None
product_type_enum: str | None = None
platform: str | None = None
google_product_category: str | None = None
category_path: str | None = None
color: str | None = None
size: str | None = None
weight: float | None = None
weight_unit: str | None = None
translations: dict | None = None
created_at: str | None = None
updated_at: str | None = None
# ============================================================================
# Endpoints
# ============================================================================
@router.get("", response_model=AdminProductListResponse)
def get_products(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=500),
search: str | None = Query(
None, description="Search by title, GTIN, SKU, or brand"
),
marketplace: str | None = Query(None, description="Filter by marketplace"),
vendor_name: str | None = Query(None, description="Filter by vendor name"),
availability: str | None = Query(None, description="Filter by availability"),
is_active: bool | None = Query(None, description="Filter by active status"),
is_digital: bool | None = Query(None, description="Filter by digital products"),
language: str = Query("en", description="Language for title lookup"),
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""
Get all marketplace products with search and filtering.
This endpoint allows admins to browse the entire product catalog
imported from various marketplaces.
"""
products, total = marketplace_product_service.get_admin_products(
db=db,
skip=skip,
limit=limit,
search=search,
marketplace=marketplace,
vendor_name=vendor_name,
availability=availability,
is_active=is_active,
is_digital=is_digital,
language=language,
)
return AdminProductListResponse(
products=[AdminProductListItem(**p) for p in products],
total=total,
skip=skip,
limit=limit,
)
@router.get("/stats", response_model=AdminProductStats)
def get_product_stats(
marketplace: str | None = Query(None, description="Filter by marketplace"),
vendor_name: str | None = Query(None, description="Filter by vendor name"),
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""Get product statistics for admin dashboard."""
stats = marketplace_product_service.get_admin_product_stats(
db, marketplace=marketplace, vendor_name=vendor_name
)
return AdminProductStats(**stats)
@router.get("/marketplaces", response_model=MarketplacesResponse)
def get_marketplaces(
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""Get list of unique marketplaces in the product catalog."""
marketplaces = marketplace_product_service.get_marketplaces_list(db)
return MarketplacesResponse(marketplaces=marketplaces)
@router.get("/vendors", response_model=VendorsResponse)
def get_product_vendors(
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""Get list of unique vendor names in the product catalog."""
vendors = marketplace_product_service.get_source_vendors_list(db)
return VendorsResponse(vendors=vendors)
@router.post("/copy-to-vendor", response_model=CopyToVendorResponse)
def copy_products_to_vendor(
request: CopyToVendorRequest,
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""
Copy marketplace products to a vendor's catalog.
This endpoint allows admins to copy products from the master marketplace
product repository to any vendor's product catalog.
The copy creates a new Product entry linked to the MarketplaceProduct,
with default values that can be overridden by the vendor later.
"""
result = marketplace_product_service.copy_to_vendor_catalog(
db=db,
marketplace_product_ids=request.marketplace_product_ids,
vendor_id=request.vendor_id,
skip_existing=request.skip_existing,
)
db.commit() # ✅ ARCH: Commit at API level for transaction control
return CopyToVendorResponse(**result)
@router.get("/{product_id}", response_model=AdminProductDetail)
def get_product_detail(
product_id: int,
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""Get detailed product information including all translations."""
product = marketplace_product_service.get_admin_product_detail(db, product_id)
return AdminProductDetail(**product)

159
app/api/v1/admin/vendor_products.py
View File

@@ -1,159 +0,0 @@
# app/api/v1/admin/vendor_products.py
"""
Admin vendor product catalog endpoints.
Provides management of vendor-specific product catalogs:
- Browse products in vendor catalogs
- View product details with override info
- Remove products from catalog
Architecture Notes:
- All Pydantic schemas are defined in models/schema/vendor_product.py
- Business logic is delegated to vendor_product_service
"""
import logging
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_api
from app.core.database import get_db
from app.services.subscription_service import subscription_service
from app.services.vendor_product_service import vendor_product_service
from models.schema.auth import UserContext
from app.modules.catalog.schemas import (
CatalogVendor,
CatalogVendorsResponse,
RemoveProductResponse,
VendorProductCreate,
VendorProductCreateResponse,
VendorProductDetail,
VendorProductListItem,
VendorProductListResponse,
VendorProductStats,
VendorProductUpdate,
)
router = APIRouter(prefix="/vendor-products")
logger = logging.getLogger(__name__)
# ============================================================================
# Endpoints
# ============================================================================
@router.get("", response_model=VendorProductListResponse)
def get_vendor_products(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=500),
search: str | None = Query(None, description="Search by title or SKU"),
vendor_id: int | None = Query(None, description="Filter by vendor"),
is_active: bool | None = Query(None, description="Filter by active status"),
is_featured: bool | None = Query(None, description="Filter by featured status"),
language: str = Query("en", description="Language for title lookup"),
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""
Get all products in vendor catalogs with filtering.
This endpoint allows admins to browse products that have been
copied to vendor catalogs from the marketplace repository.
"""
products, total = vendor_product_service.get_products(
db=db,
skip=skip,
limit=limit,
search=search,
vendor_id=vendor_id,
is_active=is_active,
is_featured=is_featured,
language=language,
)
return VendorProductListResponse(
products=[VendorProductListItem(**p) for p in products],
total=total,
skip=skip,
limit=limit,
)
@router.get("/stats", response_model=VendorProductStats)
def get_vendor_product_stats(
vendor_id: int | None = Query(None, description="Filter stats by vendor ID"),
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""Get vendor product statistics for admin dashboard."""
stats = vendor_product_service.get_product_stats(db, vendor_id=vendor_id)
return VendorProductStats(**stats)
@router.get("/vendors", response_model=CatalogVendorsResponse)
def get_catalog_vendors(
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""Get list of vendors with products in their catalogs."""
vendors = vendor_product_service.get_catalog_vendors(db)
return CatalogVendorsResponse(vendors=[CatalogVendor(**v) for v in vendors])
@router.get("/{product_id}", response_model=VendorProductDetail)
def get_vendor_product_detail(
product_id: int,
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""Get detailed vendor product information including override info."""
product = vendor_product_service.get_product_detail(db, product_id)
return VendorProductDetail(**product)
@router.post("", response_model=VendorProductCreateResponse)
def create_vendor_product(
data: VendorProductCreate,
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""Create a new vendor product."""
# Check product limit before creating
subscription_service.check_product_limit(db, data.vendor_id)
product = vendor_product_service.create_product(db, data.model_dump())
db.commit() # ✅ ARCH: Commit at API level for transaction control
return VendorProductCreateResponse(
id=product.id, message="Product created successfully"
)
@router.patch("/{product_id}", response_model=VendorProductDetail)
def update_vendor_product(
product_id: int,
data: VendorProductUpdate,
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""Update a vendor product."""
# Only include fields that were explicitly set
update_data = data.model_dump(exclude_unset=True)
vendor_product_service.update_product(db, product_id, update_data)
db.commit() # ✅ ARCH: Commit at API level for transaction control
# Return the updated product detail
product = vendor_product_service.get_product_detail(db, product_id)
return VendorProductDetail(**product)
@router.delete("/{product_id}", response_model=RemoveProductResponse)
def remove_vendor_product(
product_id: int,
db: Session = Depends(get_db),
current_admin: UserContext = Depends(get_current_admin_api),
):
"""Remove a product from vendor catalog."""
result = vendor_product_service.remove_product(db, product_id)
db.commit() # ✅ ARCH: Commit at API level for transaction control
return RemoveProductResponse(**result)