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

feat: complete marketplace module self-containment

Browse Source 
Migrate marketplace module to self-contained structure:

- routes/api/admin.py - Admin API endpoints
- routes/api/vendor.py - Vendor API endpoints
- routes/pages/ - Page routes (placeholder)
- models/letzshop.py - Letzshop model
- models/marketplace_import_job.py - Import job model
- models/marketplace_product.py - Product model
- models/marketplace_product_translation.py - Translation model
- schemas/marketplace_import_job.py - Import job schemas
- schemas/marketplace_product.py - Product schemas
- locales/ - Translations (en, de, fr, lu)

Removed legacy route files replaced by api/ structure.
Updated __init__.py files to use new structure.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-01-28 22:21:40 +01:00
parent f79e67d199
commit b74d1346aa
19 changed files with 1894 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
app/modules/marketplace/schemas/__init__.py
View File

@@ -1,9 +1,8 @@
# app/modules/marketplace/schemas/__init__.py
"""
Marketplace module Pydantic schemas.
Marketplace module Pydantic schemas for API request/response validation.
Re-exports marketplace schemas from the central schemas location.
Provides a module-local import path while maintaining backwards compatibility.
This is the canonical location for marketplace schemas.
Usage:
from app.modules.marketplace.schemas import (
@@ -13,13 +12,18 @@ Usage:
)
"""
from models.schema.marketplace_import_job import (
from app.modules.marketplace.schemas.marketplace_import_job import (
MarketplaceImportJobRequest,
AdminMarketplaceImportJobRequest,
MarketplaceImportJobResponse,
MarketplaceImportJobListResponse,
MarketplaceImportErrorResponse,
MarketplaceImportErrorListResponse,
AdminMarketplaceImportJobResponse,
AdminMarketplaceImportJobListResponse,
MarketplaceImportJobStatusUpdate,
)
from models.schema.marketplace_product import (
from app.modules.marketplace.schemas.marketplace_product import (
# Translation schemas
MarketplaceProductTranslationSchema,
# Base schemas
@@ -42,6 +46,11 @@ __all__ = [
"AdminMarketplaceImportJobRequest",
"MarketplaceImportJobResponse",
"MarketplaceImportJobListResponse",
"MarketplaceImportErrorResponse",
"MarketplaceImportErrorListResponse",
"AdminMarketplaceImportJobResponse",
"AdminMarketplaceImportJobListResponse",
"MarketplaceImportJobStatusUpdate",
# Product schemas
"MarketplaceProductTranslationSchema",
"MarketplaceProductBase",

170
app/modules/marketplace/schemas/marketplace_import_job.py Normal file
View File

@@ -0,0 +1,170 @@
# app/modules/marketplace/schemas/marketplace_import_job.py
"""Pydantic schemas for marketplace import jobs."""
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator
class MarketplaceImportJobRequest(BaseModel):
"""Request schema for triggering marketplace import.
Note: vendor_id is injected by middleware, not from request body.
"""
source_url: str = Field(..., description="URL to CSV file from marketplace")
marketplace: str = Field(default="Letzshop", description="Marketplace name")
batch_size: int | None = Field(
1000, description="Processing batch size", ge=100, le=10000
)
language: str = Field(
default="en",
description="Language code for product translations (e.g., 'en', 'fr', 'de')",
)
@field_validator("source_url")
@classmethod
def validate_url(cls, v):
if not v.startswith(("http://", "https://")): # noqa: SEC-034
raise ValueError("URL must start with http:// or https://") # noqa: SEC-034
return v.strip()
@field_validator("marketplace")
@classmethod
def validate_marketplace(cls, v):
return v.strip()
@field_validator("language")
@classmethod
def validate_language(cls, v):
# Basic language code validation (2-5 chars)
v = v.strip().lower()
if not 2 <= len(v) <= 5:
raise ValueError("Language code must be 2-5 characters (e.g., 'en', 'fr')")
return v
class AdminMarketplaceImportJobRequest(BaseModel):
"""Request schema for admin-triggered marketplace import.
Includes vendor_id since admin can import for any vendor.
"""
vendor_id: int = Field(..., description="Vendor ID to import products for")
source_url: str = Field(..., description="URL to CSV file from marketplace")
marketplace: str = Field(default="Letzshop", description="Marketplace name")
batch_size: int | None = Field(
1000, description="Processing batch size", ge=100, le=10000
)
language: str = Field(
default="en",
description="Language code for product translations (e.g., 'en', 'fr', 'de')",
)
@field_validator("source_url")
@classmethod
def validate_url(cls, v):
if not v.startswith(("http://", "https://")): # noqa: SEC-034
raise ValueError("URL must start with http:// or https://") # noqa: SEC-034
return v.strip()
@field_validator("marketplace")
@classmethod
def validate_marketplace(cls, v):
return v.strip()
@field_validator("language")
@classmethod
def validate_language(cls, v):
v = v.strip().lower()
if not 2 <= len(v) <= 5:
raise ValueError("Language code must be 2-5 characters (e.g., 'en', 'fr')")
return v
class MarketplaceImportErrorResponse(BaseModel):
"""Response schema for individual import error."""
model_config = ConfigDict(from_attributes=True)
id: int
row_number: int
identifier: str | None = None
error_type: str
error_message: str
row_data: dict | None = None
created_at: datetime
class MarketplaceImportErrorListResponse(BaseModel):
"""Response schema for list of import errors."""
errors: list[MarketplaceImportErrorResponse]
total: int
import_job_id: int
class MarketplaceImportJobResponse(BaseModel):
"""Response schema for marketplace import job."""
model_config = ConfigDict(from_attributes=True)
job_id: int
vendor_id: int
vendor_code: str | None = None # Populated from vendor relationship
vendor_name: str | None = None # Populated from vendor relationship
marketplace: str
source_url: str
status: str
language: str | None = None # Language used for translations
# Counts
imported: int = 0
updated: int = 0
total_processed: int = 0
error_count: int = 0
# Error details
error_message: str | None = None
# Timestamps
created_at: datetime
started_at: datetime | None = None
completed_at: datetime | None = None
class MarketplaceImportJobListResponse(BaseModel):
"""Response schema for list of import jobs."""
jobs: list[MarketplaceImportJobResponse]
total: int
skip: int
limit: int
class AdminMarketplaceImportJobResponse(MarketplaceImportJobResponse):
"""Extended response schema for admin with additional fields."""
id: int # Alias for job_id (frontend compatibility)
error_details: list = [] # Placeholder for future error details
created_by_name: str | None = None # Username of who created the job
class AdminMarketplaceImportJobListResponse(BaseModel):
"""Response schema for paginated list of import jobs (admin)."""
items: list[AdminMarketplaceImportJobResponse]
total: int
page: int
limit: int
class MarketplaceImportJobStatusUpdate(BaseModel):
"""Schema for updating import job status (internal use)."""
status: str
imported_count: int | None = None
updated_count: int | None = None
error_count: int | None = None
total_processed: int | None = None
error_message: str | None = None

225
app/modules/marketplace/schemas/marketplace_product.py Normal file
View File

@@ -0,0 +1,225 @@
# app/modules/marketplace/schemas/marketplace_product.py
"""Pydantic schemas for MarketplaceProduct API validation.
Note: title and description are stored in MarketplaceProductTranslation table,
but we keep them in the API schemas for convenience. The service layer
handles creating/updating translations separately.
"""
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
from models.schema.inventory import ProductInventorySummary
class MarketplaceProductTranslationSchema(BaseModel):
"""Schema for product translation."""
model_config = ConfigDict(from_attributes=True)
language: str
title: str
description: str | None = None
short_description: str | None = None
meta_title: str | None = None
meta_description: str | None = None
url_slug: str | None = None
class MarketplaceProductBase(BaseModel):
"""Base schema for marketplace products."""
marketplace_product_id: str | None = None
# Localized fields (passed to translations)
title: str | None = None
description: str | None = None
# Links and media
link: str | None = None
image_link: str | None = None
additional_image_link: str | None = None
# Status
availability: str | None = None
is_active: bool | None = None
# Pricing
price: str | None = None
sale_price: str | None = None
currency: str | None = None
# Product identifiers
brand: str | None = None
gtin: str | None = None
mpn: str | None = None
sku: str | None = None
# Product attributes
condition: str | None = None
adult: str | None = None
multipack: int | None = None
is_bundle: str | None = None
age_group: str | None = None
color: str | None = None
gender: str | None = None
material: str | None = None
pattern: str | None = None
size: str | None = None
size_type: str | None = None
size_system: str | None = None
item_group_id: str | None = None
# Categories
google_product_category: str | None = None
product_type_raw: str | None = (
None # Original feed value (renamed from product_type)
)
category_path: str | None = None
# Custom labels
custom_label_0: str | None = None
custom_label_1: str | None = None
custom_label_2: str | None = None
custom_label_3: str | None = None
custom_label_4: str | None = None
# Unit pricing
unit_pricing_measure: str | None = None
unit_pricing_base_measure: str | None = None
identifier_exists: str | None = None
shipping: str | None = None
# Source tracking
marketplace: str | None = None
vendor_name: str | None = None
source_url: str | None = None
# Product type classification
product_type_enum: str | None = (
None # 'physical', 'digital', 'service', 'subscription'
)
is_digital: bool | None = None
# Digital product fields
digital_delivery_method: str | None = None
platform: str | None = None
license_type: str | None = None
# Physical product fields
weight: float | None = None
weight_unit: str | None = None
class MarketplaceProductCreate(MarketplaceProductBase):
"""Schema for creating a marketplace product."""
marketplace_product_id: str = Field(
..., description="Unique product identifier from marketplace"
)
# Title is required for API creation (will be stored in translations)
title: str = Field(..., description="Product title")
class MarketplaceProductUpdate(MarketplaceProductBase):
"""Schema for updating a marketplace product.
All fields are optional - only provided fields will be updated.
"""
class MarketplaceProductResponse(BaseModel):
"""Schema for marketplace product API response."""
model_config = ConfigDict(from_attributes=True)
id: int
marketplace_product_id: str
# These will be populated from translations
title: str | None = None
description: str | None = None
# Links and media
link: str | None = None
image_link: str | None = None
additional_image_link: str | None = None
# Status
availability: str | None = None
is_active: bool | None = None
# Pricing
price: str | None = None
price_numeric: float | None = None
sale_price: str | None = None
sale_price_numeric: float | None = None
currency: str | None = None
# Product identifiers
brand: str | None = None
gtin: str | None = None
mpn: str | None = None
sku: str | None = None
# Product attributes
condition: str | None = None
color: str | None = None
size: str | None = None
# Categories
google_product_category: str | None = None
product_type_raw: str | None = None
category_path: str | None = None
# Source tracking
marketplace: str | None = None
vendor_name: str | None = None
# Product type
product_type_enum: str | None = None
is_digital: bool | None = None
platform: str | None = None
# Timestamps
created_at: datetime
updated_at: datetime
# Translations (optional - included when requested)
translations: list[MarketplaceProductTranslationSchema] | None = None
class MarketplaceProductListResponse(BaseModel):
"""Schema for paginated product list response."""
products: list[MarketplaceProductResponse]
total: int
skip: int
limit: int
class MarketplaceProductDetailResponse(BaseModel):
"""Schema for detailed product response with inventory."""
product: MarketplaceProductResponse
inventory_info: ProductInventorySummary | None = None
translations: list[MarketplaceProductTranslationSchema] | None = None
class MarketplaceImportRequest(BaseModel):
"""Schema for marketplace import request."""
url: str = Field(..., description="URL to CSV file")
marketplace: str = Field(default="Letzshop", description="Marketplace name")
vendor_name: str | None = Field(default=None, description="Vendor name")
language: str = Field(default="en", description="Language code for translations")
batch_size: int = Field(default=100, ge=1, le=1000, description="Batch size")
class MarketplaceImportResponse(BaseModel):
"""Schema for marketplace import response."""
job_id: int
status: str
message: str