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

feat: add Letzshop bidirectional order integration

Browse Source 
Add complete Letzshop marketplace integration with:
- GraphQL client for order import and fulfillment operations
- Encrypted credential storage per vendor (Fernet encryption)
- Admin and vendor API endpoints for credentials management
- Order import, confirmation, rejection, and tracking
- Fulfillment queue and sync logging
- Comprehensive documentation and test coverage

New files:
- app/services/letzshop/ - GraphQL client and services
- app/utils/encryption.py - Fernet encryption utility
- models/database/letzshop.py - Database models
- models/schema/letzshop.py - Pydantic schemas
- app/api/v1/admin/letzshop.py - Admin API endpoints
- app/api/v1/vendor/letzshop.py - Vendor API endpoints
- docs/guides/letzshop-order-integration.md - Documentation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2025-12-13 12:19:54 +01:00
parent 837b1f93f4
commit 448f01f82b
20 changed files with 5251 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

413
app/services/letzshop/credentials.py Normal file
View File

@@ -0,0 +1,413 @@
# app/services/letzshop/credentials.py
"""
Letzshop credentials management service.
Handles secure storage and retrieval of per-vendor Letzshop API credentials.
"""
import logging
from datetime import datetime, timezone
from sqlalchemy.orm import Session
from app.utils.encryption import decrypt_value, encrypt_value, mask_api_key
from models.database.letzshop import VendorLetzshopCredentials
from .client import LetzshopClient
logger = logging.getLogger(__name__)
# Default Letzshop GraphQL endpoint
DEFAULT_ENDPOINT = "https://letzshop.lu/graphql"
class CredentialsError(Exception):
"""Base exception for credentials errors."""
pass
class CredentialsNotFoundError(CredentialsError):
"""Raised when credentials are not found for a vendor."""
pass
class LetzshopCredentialsService:
"""
Service for managing Letzshop API credentials.
Provides secure storage and retrieval of encrypted API keys,
connection testing, and sync status updates.
"""
def __init__(self, db: Session):
"""
Initialize the credentials service.
Args:
db: SQLAlchemy database session.
"""
self.db = db
# ========================================================================
# CRUD Operations
# ========================================================================
def get_credentials(
self, vendor_id: int
) -> VendorLetzshopCredentials | None:
"""
Get Letzshop credentials for a vendor.
Args:
vendor_id: The vendor ID.
Returns:
VendorLetzshopCredentials or None if not found.
"""
return (
self.db.query(VendorLetzshopCredentials)
.filter(VendorLetzshopCredentials.vendor_id == vendor_id)
.first()
)
def get_credentials_or_raise(
self, vendor_id: int
) -> VendorLetzshopCredentials:
"""
Get Letzshop credentials for a vendor or raise an exception.
Args:
vendor_id: The vendor ID.
Returns:
VendorLetzshopCredentials.
Raises:
CredentialsNotFoundError: If credentials are not found.
"""
credentials = self.get_credentials(vendor_id)
if credentials is None:
raise CredentialsNotFoundError(
f"Letzshop credentials not found for vendor {vendor_id}"
)
return credentials
def create_credentials(
self,
vendor_id: int,
api_key: str,
api_endpoint: str | None = None,
auto_sync_enabled: bool = False,
sync_interval_minutes: int = 15,
) -> VendorLetzshopCredentials:
"""
Create Letzshop credentials for a vendor.
Args:
vendor_id: The vendor ID.
api_key: The Letzshop API key (will be encrypted).
api_endpoint: Custom API endpoint (optional).
auto_sync_enabled: Whether to enable automatic sync.
sync_interval_minutes: Sync interval in minutes.
Returns:
Created VendorLetzshopCredentials.
"""
# Encrypt the API key
encrypted_key = encrypt_value(api_key)
credentials = VendorLetzshopCredentials(
vendor_id=vendor_id,
api_key_encrypted=encrypted_key,
api_endpoint=api_endpoint or DEFAULT_ENDPOINT,
auto_sync_enabled=auto_sync_enabled,
sync_interval_minutes=sync_interval_minutes,
)
self.db.add(credentials)
self.db.commit()
self.db.refresh(credentials)
logger.info(f"Created Letzshop credentials for vendor {vendor_id}")
return credentials
def update_credentials(
self,
vendor_id: int,
api_key: str | None = None,
api_endpoint: str | None = None,
auto_sync_enabled: bool | None = None,
sync_interval_minutes: int | None = None,
) -> VendorLetzshopCredentials:
"""
Update Letzshop credentials for a vendor.
Args:
vendor_id: The vendor ID.
api_key: New API key (optional, will be encrypted if provided).
api_endpoint: New API endpoint (optional).
auto_sync_enabled: New auto-sync setting (optional).
sync_interval_minutes: New sync interval (optional).
Returns:
Updated VendorLetzshopCredentials.
Raises:
CredentialsNotFoundError: If credentials are not found.
"""
credentials = self.get_credentials_or_raise(vendor_id)
if api_key is not None:
credentials.api_key_encrypted = encrypt_value(api_key)
if api_endpoint is not None:
credentials.api_endpoint = api_endpoint
if auto_sync_enabled is not None:
credentials.auto_sync_enabled = auto_sync_enabled
if sync_interval_minutes is not None:
credentials.sync_interval_minutes = sync_interval_minutes
self.db.commit()
self.db.refresh(credentials)
logger.info(f"Updated Letzshop credentials for vendor {vendor_id}")
return credentials
def delete_credentials(self, vendor_id: int) -> bool:
"""
Delete Letzshop credentials for a vendor.
Args:
vendor_id: The vendor ID.
Returns:
True if deleted, False if not found.
"""
credentials = self.get_credentials(vendor_id)
if credentials is None:
return False
self.db.delete(credentials)
self.db.commit()
logger.info(f"Deleted Letzshop credentials for vendor {vendor_id}")
return True
def upsert_credentials(
self,
vendor_id: int,
api_key: str,
api_endpoint: str | None = None,
auto_sync_enabled: bool = False,
sync_interval_minutes: int = 15,
) -> VendorLetzshopCredentials:
"""
Create or update Letzshop credentials for a vendor.
Args:
vendor_id: The vendor ID.
api_key: The Letzshop API key (will be encrypted).
api_endpoint: Custom API endpoint (optional).
auto_sync_enabled: Whether to enable automatic sync.
sync_interval_minutes: Sync interval in minutes.
Returns:
Created or updated VendorLetzshopCredentials.
"""
existing = self.get_credentials(vendor_id)
if existing:
return self.update_credentials(
vendor_id=vendor_id,
api_key=api_key,
api_endpoint=api_endpoint,
auto_sync_enabled=auto_sync_enabled,
sync_interval_minutes=sync_interval_minutes,
)
return self.create_credentials(
vendor_id=vendor_id,
api_key=api_key,
api_endpoint=api_endpoint,
auto_sync_enabled=auto_sync_enabled,
sync_interval_minutes=sync_interval_minutes,
)
# ========================================================================
# Key Decryption and Client Creation
# ========================================================================
def get_decrypted_api_key(self, vendor_id: int) -> str:
"""
Get the decrypted API key for a vendor.
Args:
vendor_id: The vendor ID.
Returns:
Decrypted API key.
Raises:
CredentialsNotFoundError: If credentials are not found.
"""
credentials = self.get_credentials_or_raise(vendor_id)
return decrypt_value(credentials.api_key_encrypted)
def get_masked_api_key(self, vendor_id: int) -> str:
"""
Get a masked version of the API key for display.
Args:
vendor_id: The vendor ID.
Returns:
Masked API key (e.g., "sk-a***************").
Raises:
CredentialsNotFoundError: If credentials are not found.
"""
api_key = self.get_decrypted_api_key(vendor_id)
return mask_api_key(api_key)
def create_client(self, vendor_id: int) -> LetzshopClient:
"""
Create a Letzshop client for a vendor.
Args:
vendor_id: The vendor ID.
Returns:
Configured LetzshopClient.
Raises:
CredentialsNotFoundError: If credentials are not found.
"""
credentials = self.get_credentials_or_raise(vendor_id)
api_key = decrypt_value(credentials.api_key_encrypted)
return LetzshopClient(
api_key=api_key,
endpoint=credentials.api_endpoint,
)
# ========================================================================
# Connection Testing
# ========================================================================
def test_connection(
self, vendor_id: int
) -> tuple[bool, float | None, str | None]:
"""
Test the connection for a vendor's credentials.
Args:
vendor_id: The vendor ID.
Returns:
Tuple of (success, response_time_ms, error_message).
"""
try:
with self.create_client(vendor_id) as client:
return client.test_connection()
except CredentialsNotFoundError:
return False, None, "Letzshop credentials not configured"
except Exception as e:
logger.error(f"Connection test failed for vendor {vendor_id}: {e}")
return False, None, str(e)
def test_api_key(
self,
api_key: str,
api_endpoint: str | None = None,
) -> tuple[bool, float | None, str | None]:
"""
Test an API key without saving it.
Args:
api_key: The API key to test.
api_endpoint: Optional custom endpoint.
Returns:
Tuple of (success, response_time_ms, error_message).
"""
try:
with LetzshopClient(
api_key=api_key,
endpoint=api_endpoint or DEFAULT_ENDPOINT,
) as client:
return client.test_connection()
except Exception as e:
logger.error(f"API key test failed: {e}")
return False, None, str(e)
# ========================================================================
# Sync Status Updates
# ========================================================================
def update_sync_status(
self,
vendor_id: int,
status: str,
error: str | None = None,
) -> VendorLetzshopCredentials | None:
"""
Update the last sync status for a vendor.
Args:
vendor_id: The vendor ID.
status: Sync status (success, failed, partial).
error: Error message if sync failed.
Returns:
Updated credentials or None if not found.
"""
credentials = self.get_credentials(vendor_id)
if credentials is None:
return None
credentials.last_sync_at = datetime.now(timezone.utc)
credentials.last_sync_status = status
credentials.last_sync_error = error
self.db.commit()
self.db.refresh(credentials)
return credentials
# ========================================================================
# Status Helpers
# ========================================================================
def is_configured(self, vendor_id: int) -> bool:
"""Check if Letzshop is configured for a vendor."""
return self.get_credentials(vendor_id) is not None
def get_status(self, vendor_id: int) -> dict:
"""
Get the Letzshop integration status for a vendor.
Args:
vendor_id: The vendor ID.
Returns:
Status dictionary with configuration and sync info.
"""
credentials = self.get_credentials(vendor_id)
if credentials is None:
return {
"is_configured": False,
"is_connected": False,
"last_sync_at": None,
"last_sync_status": None,
"auto_sync_enabled": False,
}
return {
"is_configured": True,
"is_connected": credentials.last_sync_status == "success",
"last_sync_at": credentials.last_sync_at,
"last_sync_status": credentials.last_sync_status,
"auto_sync_enabled": credentials.auto_sync_enabled,
}