sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8dcc4145aa3d996724eae48192b7433fa5d3fd47
orion/app/modules/loyalty/routes/api/store.py
Samir Boulahtit 7c43d6f4a2 refactor: fix all architecture validator findings (202 → 0) 
Eliminate all 103 errors and 96 warnings from the architecture validator:

Phase 1 - Validator rules & YAML:
- Add NAM-001/NAM-002 exceptions for module-scoped router/service files
- Fix API-004 to detect # public comments on decorator lines
- Add module-specific exception bases to EXC-004 valid_bases
- Exclude storefront files from AUTH-004 store context check
- Add SVC-006 exceptions for loyalty service atomic commits
- Fix _get_rule() to search naming_rules and auth_rules categories
- Use plain # CODE comments instead of # noqa: CODE for custom rules

Phase 2 - Billing module (5 route files):
- Move _resolve_store_to_merchant to subscription_service
- Move tier/feature queries to feature_service, admin_subscription_service
- Extract 22 inline Pydantic schemas to billing/schemas/billing.py
- Replace all HTTPException with domain exceptions

Phase 3 - Loyalty module (4 routes + points_service):
- Add 7 domain exceptions (Apple auth, enrollment, device registration)
- Add service methods to card_service, program_service, apple_wallet_service
- Move all db.query() from routes to service layer
- Fix SVC-001: replace HTTPException in points_service with domain exception

Phase 4 - Remaining modules:
- tenancy: move store stats queries to admin_service
- cms: move platform resolution to content_page_service, add NoPlatformSubscriptionException
- messaging: move user/customer lookups to messaging_service
- Add ConfigDict(from_attributes=True) to ContentPageResponse

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-13 18:49:24 +01:00

649 lines
20 KiB
Python
Raw Blame History

# app/modules/loyalty/routes/api/store.py
"""
Loyalty module store routes.
Merchant-based store endpoints for:
- Program management (merchant-wide, managed by store)
- Staff PINs (per-store)
- Card operations (stamps, points, redemptions, voids)
- Customer cards lookup
- Dashboard stats
All operations are scoped to the store's merchant.
Cards can be used at any store within the same merchant.
"""
import logging
from fastapi import APIRouter, Depends, Path, Query, Request
from sqlalchemy.orm import Session
from app.api.deps import get_current_store_api, require_module_access
from app.core.database import get_db
from app.modules.enums import FrontendType
from app.modules.loyalty.schemas import (
CardEnrollRequest,
CardListResponse,
CardLookupResponse,
CardResponse,
MerchantStatsResponse,
PinCreate,
PinListResponse,
PinResponse,
PinUpdate,
PointsAdjustRequest,
PointsAdjustResponse,
PointsEarnRequest,
PointsEarnResponse,
PointsRedeemRequest,
PointsRedeemResponse,
PointsVoidRequest,
PointsVoidResponse,
ProgramCreate,
ProgramResponse,
ProgramStatsResponse,
ProgramUpdate,
StampRedeemRequest,
StampRedeemResponse,
StampRequest,
StampResponse,
StampVoidRequest,
StampVoidResponse,
TransactionListResponse,
TransactionResponse,
)
from app.modules.loyalty.services import (
card_service,
pin_service,
points_service,
program_service,
stamp_service,
)
from app.modules.tenancy.models import User # API-007
logger = logging.getLogger(__name__)
# Store router with module access control
store_router = APIRouter(
prefix="/loyalty",
dependencies=[Depends(require_module_access("loyalty", FrontendType.STORE))],
)
def get_client_info(request: Request) -> tuple[str | None, str | None]:
"""Extract client IP and user agent from request."""
ip = request.client.host if request.client else None
user_agent = request.headers.get("user-agent")
return ip, user_agent
def get_store_merchant_id(db: Session, store_id: int) -> int:
"""Get the merchant ID for a store."""
return program_service.get_store_merchant_id(db, store_id)
# =============================================================================
# Program Management
# =============================================================================
@store_router.get("/program", response_model=ProgramResponse)
def get_program(
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Get the merchant's loyalty program."""
store_id = current_user.token_store_id
program = program_service.require_program_by_store(db, store_id)
response = ProgramResponse.model_validate(program)
response.is_stamps_enabled = program.is_stamps_enabled
response.is_points_enabled = program.is_points_enabled
response.display_name = program.display_name
return response
@store_router.post("/program", response_model=ProgramResponse, status_code=201)
def create_program(
data: ProgramCreate,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Create a loyalty program for the merchant."""
store_id = current_user.token_store_id
merchant_id = get_store_merchant_id(db, store_id)
program = program_service.create_program(db, merchant_id, data)
response = ProgramResponse.model_validate(program)
response.is_stamps_enabled = program.is_stamps_enabled
response.is_points_enabled = program.is_points_enabled
response.display_name = program.display_name
return response
@store_router.patch("/program", response_model=ProgramResponse)
def update_program(
data: ProgramUpdate,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Update the merchant's loyalty program."""
store_id = current_user.token_store_id
program = program_service.require_program_by_store(db, store_id)
program = program_service.update_program(db, program.id, data)
response = ProgramResponse.model_validate(program)
response.is_stamps_enabled = program.is_stamps_enabled
response.is_points_enabled = program.is_points_enabled
response.display_name = program.display_name
return response
@store_router.get("/stats", response_model=ProgramStatsResponse)
def get_stats(
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Get loyalty program statistics."""
store_id = current_user.token_store_id
program = program_service.require_program_by_store(db, store_id)
stats = program_service.get_program_stats(db, program.id)
return ProgramStatsResponse(**stats)
@store_router.get("/stats/merchant", response_model=MerchantStatsResponse)
def get_merchant_stats(
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Get merchant-wide loyalty statistics across all locations."""
store_id = current_user.token_store_id
merchant_id = get_store_merchant_id(db, store_id)
stats = program_service.get_merchant_stats(db, merchant_id)
return MerchantStatsResponse(**stats)
# =============================================================================
# Staff PINs
# =============================================================================
@store_router.get("/pins", response_model=PinListResponse)
def list_pins(
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""List staff PINs for this store location."""
store_id = current_user.token_store_id
program = program_service.require_program_by_store(db, store_id)
# List PINs for this store only
pins = pin_service.list_pins(db, program.id, store_id=store_id)
return PinListResponse(
pins=[PinResponse.model_validate(pin) for pin in pins],
total=len(pins),
)
@store_router.post("/pins", response_model=PinResponse, status_code=201)
def create_pin(
data: PinCreate,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Create a new staff PIN for this store location."""
store_id = current_user.token_store_id
program = program_service.require_program_by_store(db, store_id)
pin = pin_service.create_pin(db, program.id, store_id, data)
return PinResponse.model_validate(pin)
@store_router.patch("/pins/{pin_id}", response_model=PinResponse)
def update_pin(
pin_id: int = Path(..., gt=0),
data: PinUpdate = None,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Update a staff PIN."""
pin = pin_service.update_pin(db, pin_id, data)
return PinResponse.model_validate(pin)
@store_router.delete("/pins/{pin_id}", status_code=204)
def delete_pin(
pin_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Delete a staff PIN."""
pin_service.delete_pin(db, pin_id)
@store_router.post("/pins/{pin_id}/unlock", response_model=PinResponse)
def unlock_pin(
pin_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Unlock a locked staff PIN."""
pin = pin_service.unlock_pin(db, pin_id)
return PinResponse.model_validate(pin)
# =============================================================================
# Card Management
# =============================================================================
@store_router.get("/cards", response_model=CardListResponse)
def list_cards(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
is_active: bool | None = Query(None),
search: str | None = Query(None, max_length=100),
enrolled_here: bool = Query(False, description="Only show cards enrolled at this location"),
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""
List loyalty cards for the merchant.
By default lists all cards in the merchant's loyalty program.
Use enrolled_here=true to filter to cards enrolled at this location.
"""
store_id = current_user.token_store_id
merchant_id = get_store_merchant_id(db, store_id)
program = program_service.require_program_by_store(db, store_id)
# Filter by enrolled_at_store_id if requested
filter_store_id = store_id if enrolled_here else None
cards, total = card_service.list_cards(
db,
merchant_id,
store_id=filter_store_id,
skip=skip,
limit=limit,
is_active=is_active,
search=search,
)
card_responses = []
for card in cards:
response = CardResponse(
id=card.id,
card_number=card.card_number,
customer_id=card.customer_id,
merchant_id=card.merchant_id,
program_id=card.program_id,
enrolled_at_store_id=card.enrolled_at_store_id,
stamp_count=card.stamp_count,
stamps_target=program.stamps_target,
stamps_until_reward=max(0, program.stamps_target - card.stamp_count),
total_stamps_earned=card.total_stamps_earned,
stamps_redeemed=card.stamps_redeemed,
points_balance=card.points_balance,
total_points_earned=card.total_points_earned,
points_redeemed=card.points_redeemed,
is_active=card.is_active,
created_at=card.created_at,
has_google_wallet=bool(card.google_object_id),
has_apple_wallet=bool(card.apple_serial_number),
)
card_responses.append(response)
return CardListResponse(cards=card_responses, total=total)
@store_router.post("/cards/lookup", response_model=CardLookupResponse)
def lookup_card(
request: Request,
card_id: int | None = Query(None),
qr_code: str | None = Query(None),
card_number: str | None = Query(None),
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""
Look up a card by ID, QR code, or card number.
Card must belong to the same merchant as the store.
"""
store_id = current_user.token_store_id
# Uses lookup_card_for_store which validates merchant membership
# Raises LoyaltyCardNotFoundException if not found
card = card_service.lookup_card_for_store(
db,
store_id,
card_id=card_id,
qr_code=qr_code,
card_number=card_number,
)
program = card.program
# Check cooldown
can_stamp, _ = card.can_stamp(program.cooldown_minutes)
cooldown_ends = None
if not can_stamp and card.last_stamp_at:
from datetime import timedelta
cooldown_ends = card.last_stamp_at + timedelta(minutes=program.cooldown_minutes)
# Get stamps today
stamps_today = card_service.get_stamps_today(db, card.id)
# Get available points rewards
available_rewards = []
for reward in program.points_rewards or []:
if reward.get("is_active", True) and card.points_balance >= reward.get("points_required", 0):
available_rewards.append(reward)
return CardLookupResponse(
card_id=card.id,
card_number=card.card_number,
customer_id=card.customer_id,
customer_name=card.customer.full_name if card.customer else None,
customer_email=card.customer.email if card.customer else "",
merchant_id=card.merchant_id,
merchant_name=card.merchant.name if card.merchant else None,
stamp_count=card.stamp_count,
stamps_target=program.stamps_target,
stamps_until_reward=max(0, program.stamps_target - card.stamp_count),
points_balance=card.points_balance,
can_redeem_stamps=card.stamp_count >= program.stamps_target,
stamp_reward_description=program.stamps_reward_description,
available_rewards=available_rewards,
can_stamp=can_stamp,
cooldown_ends_at=cooldown_ends,
stamps_today=stamps_today,
max_daily_stamps=program.max_daily_stamps,
can_earn_more_stamps=stamps_today < program.max_daily_stamps,
)
@store_router.post("/cards/enroll", response_model=CardResponse, status_code=201)
def enroll_customer(
data: CardEnrollRequest,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""
Enroll a customer in the merchant's loyalty program.
The card will be associated with the merchant and track which
store enrolled them.
"""
store_id = current_user.token_store_id
customer_id = card_service.resolve_customer_id(
db,
customer_id=data.customer_id,
email=data.email,
store_id=store_id,
)
card = card_service.enroll_customer_for_store(db, customer_id, store_id)
program = card.program
return CardResponse(
id=card.id,
card_number=card.card_number,
customer_id=card.customer_id,
merchant_id=card.merchant_id,
program_id=card.program_id,
enrolled_at_store_id=card.enrolled_at_store_id,
stamp_count=card.stamp_count,
stamps_target=program.stamps_target,
stamps_until_reward=max(0, program.stamps_target - card.stamp_count),
total_stamps_earned=card.total_stamps_earned,
stamps_redeemed=card.stamps_redeemed,
points_balance=card.points_balance,
total_points_earned=card.total_points_earned,
points_redeemed=card.points_redeemed,
is_active=card.is_active,
created_at=card.created_at,
)
@store_router.get("/cards/{card_id}/transactions", response_model=TransactionListResponse)
def get_card_transactions(
card_id: int = Path(..., gt=0),
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Get transaction history for a card."""
store_id = current_user.token_store_id
# Verify card belongs to this merchant (raises LoyaltyCardNotFoundException if not found)
card_service.lookup_card_for_store(db, store_id, card_id=card_id)
transactions, total = card_service.get_card_transactions(
db, card_id, skip=skip, limit=limit
)
return TransactionListResponse(
transactions=[TransactionResponse.model_validate(t) for t in transactions],
total=total,
)
# =============================================================================
# Stamp Operations
# =============================================================================
@store_router.post("/stamp", response_model=StampResponse)
def add_stamp(
request: Request,
data: StampRequest,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Add a stamp to a loyalty card."""
store_id = current_user.token_store_id
ip, user_agent = get_client_info(request)
result = stamp_service.add_stamp(
db,
store_id=store_id,
card_id=data.card_id,
qr_code=data.qr_code,
card_number=data.card_number,
staff_pin=data.staff_pin,
ip_address=ip,
user_agent=user_agent,
notes=data.notes,
)
return StampResponse(**result)
@store_router.post("/stamp/redeem", response_model=StampRedeemResponse)
def redeem_stamps(
request: Request,
data: StampRedeemRequest,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Redeem stamps for a reward."""
store_id = current_user.token_store_id
ip, user_agent = get_client_info(request)
result = stamp_service.redeem_stamps(
db,
store_id=store_id,
card_id=data.card_id,
qr_code=data.qr_code,
card_number=data.card_number,
staff_pin=data.staff_pin,
ip_address=ip,
user_agent=user_agent,
notes=data.notes,
)
return StampRedeemResponse(**result)
@store_router.post("/stamp/void", response_model=StampVoidResponse)
def void_stamps(
request: Request,
data: StampVoidRequest,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Void stamps for a return."""
store_id = current_user.token_store_id
ip, user_agent = get_client_info(request)
result = stamp_service.void_stamps(
db,
store_id=store_id,
card_id=data.card_id,
qr_code=data.qr_code,
card_number=data.card_number,
stamps_to_void=data.stamps_to_void,
original_transaction_id=data.original_transaction_id,
staff_pin=data.staff_pin,
ip_address=ip,
user_agent=user_agent,
notes=data.notes,
)
return StampVoidResponse(**result)
# =============================================================================
# Points Operations
# =============================================================================
@store_router.post("/points", response_model=PointsEarnResponse)
def earn_points(
request: Request,
data: PointsEarnRequest,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Earn points from a purchase."""
store_id = current_user.token_store_id
ip, user_agent = get_client_info(request)
result = points_service.earn_points(
db,
store_id=store_id,
card_id=data.card_id,
qr_code=data.qr_code,
card_number=data.card_number,
purchase_amount_cents=data.purchase_amount_cents,
order_reference=data.order_reference,
staff_pin=data.staff_pin,
ip_address=ip,
user_agent=user_agent,
notes=data.notes,
)
return PointsEarnResponse(**result)
@store_router.post("/points/redeem", response_model=PointsRedeemResponse)
def redeem_points(
request: Request,
data: PointsRedeemRequest,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Redeem points for a reward."""
store_id = current_user.token_store_id
ip, user_agent = get_client_info(request)
result = points_service.redeem_points(
db,
store_id=store_id,
card_id=data.card_id,
qr_code=data.qr_code,
card_number=data.card_number,
reward_id=data.reward_id,
staff_pin=data.staff_pin,
ip_address=ip,
user_agent=user_agent,
notes=data.notes,
)
return PointsRedeemResponse(**result)
@store_router.post("/points/void", response_model=PointsVoidResponse)
def void_points(
request: Request,
data: PointsVoidRequest,
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Void points for a return."""
store_id = current_user.token_store_id
ip, user_agent = get_client_info(request)
result = points_service.void_points(
db,
store_id=store_id,
card_id=data.card_id,
qr_code=data.qr_code,
card_number=data.card_number,
points_to_void=data.points_to_void,
original_transaction_id=data.original_transaction_id,
order_reference=data.order_reference,
staff_pin=data.staff_pin,
ip_address=ip,
user_agent=user_agent,
notes=data.notes,
)
return PointsVoidResponse(**result)
@store_router.post("/cards/{card_id}/points/adjust", response_model=PointsAdjustResponse)
def adjust_points(
request: Request,
data: PointsAdjustRequest,
card_id: int = Path(..., gt=0),
current_user: User = Depends(get_current_store_api),
db: Session = Depends(get_db),
):
"""Manually adjust points (store operation)."""
store_id = current_user.token_store_id
ip, user_agent = get_client_info(request)
result = points_service.adjust_points(
db,
card_id=card_id,
points_delta=data.points_delta,
store_id=store_id,
reason=data.reason,
staff_pin=data.staff_pin,
ip_address=ip,
user_agent=user_agent,
)
return PointsAdjustResponse(**result)
Reference in New Issue View Git Blame Copy Permalink