feat: add partial shipment support (Phase 3)

- Add shipped_quantity field to OrderItem for tracking partial fulfillment
- Add partially_shipped order status for orders with partial shipments
- Add fulfill_item method for shipping individual items with quantities
- Add get_shipment_status method for detailed shipment tracking
- Add vendor API endpoints for partial shipment operations:
  - GET /orders/{id}/shipment-status - Get item-level shipment status
  - POST /orders/{id}/items/{item_id}/ship - Ship specific item quantity
- Automatic status updates: partially_shipped when some items shipped,
  shipped when all items fully shipped
- Migration to add shipped_quantity column with upgrade for existing data
- Update documentation with partial shipment usage examples

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

alembic/versions/p4d5e6f7a8b9_add_shipped_quantity_to_order_items.py

@@ -0,0 +1,39 @@
+# alembic/versions/p4d5e6f7a8b9_add_shipped_quantity_to_order_items.py
+"""Add shipped_quantity to order_items for partial shipments.
+
+Revision ID: p4d5e6f7a8b9
+Revises: o3c4d5e6f7a8
+Create Date: 2026-01-01 12:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'p4d5e6f7a8b9'
+down_revision: Union[str, None] = 'o3c4d5e6f7a8'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Add shipped_quantity column to order_items
+    op.add_column(
+        'order_items',
+        sa.Column('shipped_quantity', sa.Integer(), nullable=False, server_default='0')
+    )
+
+    # Set shipped_quantity = quantity for already fulfilled items
+    # This handles existing data where inventory_fulfilled is True
+    op.execute("""
+        UPDATE order_items
+        SET shipped_quantity = quantity
+        WHERE inventory_fulfilled = 1
+    """)
+
+
+def downgrade() -> None:
+    op.drop_column('order_items', 'shipped_quantity')

app/api/v1/vendor/orders.py

@@ -9,10 +9,12 @@ The get_current_vendor_api dependency guarantees token_vendor_id is present.
 import logging
 
 from fastapi import APIRouter, Depends, Query
+from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 
 from app.api.deps import get_current_vendor_api
 from app.core.database import get_db
+from app.services.order_inventory_service import order_inventory_service
 from app.services.order_service import order_service
 from models.database.user import User
 from models.schema.order import (
@@ -114,3 +116,154 @@ def update_order_status(
     )
 
     return OrderResponse.model_validate(order)
+
+
+# ============================================================================
+# Partial Shipment Endpoints
+# ============================================================================
+
+
+class ShipItemRequest(BaseModel):
+    """Request to ship specific quantity of an order item."""
+
+    quantity: int | None = Field(
+        None, ge=1, description="Quantity to ship (default: remaining quantity)"
+    )
+
+
+class ShipItemResponse(BaseModel):
+    """Response from shipping an item."""
+
+    order_id: int
+    item_id: int
+    fulfilled_quantity: int
+    shipped_quantity: int | None = None
+    remaining_quantity: int | None = None
+    is_fully_shipped: bool | None = None
+    message: str | None = None
+
+
+class ShipmentStatusItemResponse(BaseModel):
+    """Item-level shipment status."""
+
+    item_id: int
+    product_id: int
+    product_name: str
+    quantity: int
+    shipped_quantity: int
+    remaining_quantity: int
+    is_fully_shipped: bool
+    is_partially_shipped: bool
+
+
+class ShipmentStatusResponse(BaseModel):
+    """Order shipment status response."""
+
+    order_id: int
+    order_number: str
+    order_status: str
+    is_fully_shipped: bool
+    is_partially_shipped: bool
+    shipped_item_count: int
+    total_item_count: int
+    total_shipped_units: int
+    total_ordered_units: int
+    items: list[ShipmentStatusItemResponse]
+
+
+@router.get("/{order_id}/shipment-status", response_model=ShipmentStatusResponse)
+def get_shipment_status(
+    order_id: int,
+    current_user: User = Depends(get_current_vendor_api),
+    db: Session = Depends(get_db),
+):
+    """
+    Get detailed shipment status for an order.
+
+    Returns item-level shipment status showing what has been shipped
+    and what remains. Useful for partial shipment tracking.
+
+    Requires Authorization header (API endpoint).
+    """
+    result = order_inventory_service.get_shipment_status(
+        db=db,
+        vendor_id=current_user.token_vendor_id,
+        order_id=order_id,
+    )
+
+    return ShipmentStatusResponse(
+        order_id=result["order_id"],
+        order_number=result["order_number"],
+        order_status=result["order_status"],
+        is_fully_shipped=result["is_fully_shipped"],
+        is_partially_shipped=result["is_partially_shipped"],
+        shipped_item_count=result["shipped_item_count"],
+        total_item_count=result["total_item_count"],
+        total_shipped_units=result["total_shipped_units"],
+        total_ordered_units=result["total_ordered_units"],
+        items=[ShipmentStatusItemResponse(**item) for item in result["items"]],
+    )
+
+
+@router.post("/{order_id}/items/{item_id}/ship", response_model=ShipItemResponse)
+def ship_order_item(
+    order_id: int,
+    item_id: int,
+    request: ShipItemRequest | None = None,
+    current_user: User = Depends(get_current_vendor_api),
+    db: Session = Depends(get_db),
+):
+    """
+    Ship a specific order item (supports partial shipment).
+
+    Fulfills inventory and updates the item's shipped quantity.
+    If quantity is not specified, ships the remaining quantity.
+
+    Example use cases:
+    - Ship all of an item: POST /orders/{id}/items/{item_id}/ship
+    - Ship partial: POST /orders/{id}/items/{item_id}/ship with {"quantity": 2}
+
+    Requires Authorization header (API endpoint).
+    """
+    quantity = request.quantity if request else None
+
+    result = order_inventory_service.fulfill_item(
+        db=db,
+        vendor_id=current_user.token_vendor_id,
+        order_id=order_id,
+        item_id=item_id,
+        quantity=quantity,
+        skip_missing=True,
+    )
+
+    # Update order status based on shipment state
+    order = order_service.get_order(db, current_user.token_vendor_id, order_id)
+
+    if order.is_fully_shipped and order.status != "shipped":
+        order_service.update_order_status(
+            db=db,
+            vendor_id=current_user.token_vendor_id,
+            order_id=order_id,
+            order_update=OrderUpdate(status="shipped"),
+        )
+        logger.info(f"Order {order.order_number} fully shipped")
+    elif order.is_partially_shipped and order.status not in (
+        "partially_shipped",
+        "shipped",
+    ):
+        order_service.update_order_status(
+            db=db,
+            vendor_id=current_user.token_vendor_id,
+            order_id=order_id,
+            order_update=OrderUpdate(status="partially_shipped"),
+        )
+        logger.info(f"Order {order.order_number} partially shipped")
+
+    db.commit()
+
+    logger.info(
+        f"Shipped item {item_id} of order {order_id}: "
+        f"{result.get('fulfilled_quantity', 0)} units"
+    )
+
+    return ShipItemResponse(**result)

app/services/order_inventory_service.py

@@ -261,6 +261,10 @@ class OrderInventoryService:
         skipped_items = []
 
         for item in order.items:
+            # Skip already fully shipped items
+            if item.is_fully_shipped:
+                continue
+
             # Skip placeholder products
             if self._is_placeholder_product(item):
                 skipped_items.append({
@@ -269,6 +273,9 @@ class OrderInventoryService:
                 })
                 continue
 
+            # Only fulfill remaining quantity
+            quantity_to_fulfill = item.remaining_quantity
+
             # Find inventory location
             location = self._find_inventory_location(db, item.product_id, vendor_id)
 
@@ -302,13 +309,17 @@ class OrderInventoryService:
                 reserve_data = InventoryReserve(
                     product_id=item.product_id,
                     location=location,
-                    quantity=item.quantity,
+                    quantity=quantity_to_fulfill,
                 )
                 updated_inventory = inventory_service.fulfill_reservation(
                     db, vendor_id, reserve_data
                 )
                 fulfilled_count += 1
 
+                # Update item shipped quantity
+                item.shipped_quantity = item.quantity
+                item.inventory_fulfilled = True
+
                 # Log transaction for audit trail
                 self._log_transaction(
                     db=db,
@@ -316,13 +327,13 @@ class OrderInventoryService:
                     product_id=item.product_id,
                     inventory=updated_inventory,
                     transaction_type=TransactionType.FULFILL,
-                    quantity_change=-item.quantity,  # Negative because stock is consumed
+                    quantity_change=-quantity_to_fulfill,  # Negative because stock is consumed
                     order=order,
                     reason=f"Fulfilled for order {order.order_number}",
                 )
 
                 logger.info(
-                    f"Fulfilled {item.quantity} units of product {item.product_id} "
+                    f"Fulfilled {quantity_to_fulfill} units of product {item.product_id} "
                     f"for order {order.order_number}"
                 )
             except (InsufficientInventoryException, InventoryNotFoundException) as e:
@@ -347,6 +358,163 @@ class OrderInventoryService:
             "skipped_items": skipped_items,
         }
 
+    def fulfill_item(
+        self,
+        db: Session,
+        vendor_id: int,
+        order_id: int,
+        item_id: int,
+        quantity: int | None = None,
+        skip_missing: bool = True,
+    ) -> dict:
+        """
+        Fulfill (deduct) inventory for a specific order item.
+
+        Supports partial fulfillment - ship some units now, rest later.
+
+        Args:
+            db: Database session
+            vendor_id: Vendor ID
+            order_id: Order ID
+            item_id: Order item ID
+            quantity: Quantity to ship (defaults to remaining quantity)
+            skip_missing: If True, skip if inventory not found
+
+        Returns:
+            Dict with fulfillment result
+
+        Raises:
+            ValidationException: If quantity exceeds remaining
+        """
+        order = self.get_order_with_items(db, vendor_id, order_id)
+
+        # Find the item
+        item = None
+        for order_item in order.items:
+            if order_item.id == item_id:
+                item = order_item
+                break
+
+        if not item:
+            raise ValidationException(f"Item {item_id} not found in order {order_id}")
+
+        # Check if already fully shipped
+        if item.is_fully_shipped:
+            return {
+                "order_id": order_id,
+                "item_id": item_id,
+                "fulfilled_quantity": 0,
+                "message": "Item already fully shipped",
+            }
+
+        # Default to remaining quantity
+        quantity_to_fulfill = quantity or item.remaining_quantity
+
+        # Validate quantity
+        if quantity_to_fulfill > item.remaining_quantity:
+            raise ValidationException(
+                f"Cannot ship {quantity_to_fulfill} units - only {item.remaining_quantity} remaining"
+            )
+
+        if quantity_to_fulfill <= 0:
+            return {
+                "order_id": order_id,
+                "item_id": item_id,
+                "fulfilled_quantity": 0,
+                "message": "Nothing to fulfill",
+            }
+
+        # Skip placeholder products
+        if self._is_placeholder_product(item):
+            return {
+                "order_id": order_id,
+                "item_id": item_id,
+                "fulfilled_quantity": 0,
+                "message": "Placeholder product - skipped",
+            }
+
+        # Find inventory location
+        location = self._find_inventory_location(db, item.product_id, vendor_id)
+
+        if not location:
+            inventory = (
+                db.query(Inventory)
+                .filter(
+                    Inventory.product_id == item.product_id,
+                    Inventory.vendor_id == vendor_id,
+                )
+                .first()
+            )
+            if inventory:
+                location = inventory.location
+
+        if not location:
+            if skip_missing:
+                return {
+                    "order_id": order_id,
+                    "item_id": item_id,
+                    "fulfilled_quantity": 0,
+                    "message": "No inventory found",
+                }
+            else:
+                raise InventoryNotFoundException(
+                    f"No inventory found for product {item.product_id}"
+                )
+
+        try:
+            reserve_data = InventoryReserve(
+                product_id=item.product_id,
+                location=location,
+                quantity=quantity_to_fulfill,
+            )
+            updated_inventory = inventory_service.fulfill_reservation(
+                db, vendor_id, reserve_data
+            )
+
+            # Update item shipped quantity
+            item.shipped_quantity += quantity_to_fulfill
+
+            # Mark as fulfilled only if fully shipped
+            if item.is_fully_shipped:
+                item.inventory_fulfilled = True
+
+            # Log transaction
+            self._log_transaction(
+                db=db,
+                vendor_id=vendor_id,
+                product_id=item.product_id,
+                inventory=updated_inventory,
+                transaction_type=TransactionType.FULFILL,
+                quantity_change=-quantity_to_fulfill,
+                order=order,
+                reason=f"Partial shipment for order {order.order_number}",
+            )
+
+            logger.info(
+                f"Fulfilled {quantity_to_fulfill} of {item.quantity} units "
+                f"for item {item_id} in order {order.order_number}"
+            )
+
+            return {
+                "order_id": order_id,
+                "item_id": item_id,
+                "fulfilled_quantity": quantity_to_fulfill,
+                "shipped_quantity": item.shipped_quantity,
+                "remaining_quantity": item.remaining_quantity,
+                "is_fully_shipped": item.is_fully_shipped,
+            }
+
+        except (InsufficientInventoryException, InventoryNotFoundException) as e:
+            if skip_missing:
+                return {
+                    "order_id": order_id,
+                    "item_id": item_id,
+                    "fulfilled_quantity": 0,
+                    "message": str(e),
+                }
+            else:
+                raise
+
     def release_order_reservation(
         self,
         db: Session,
@@ -468,6 +636,7 @@ class OrderInventoryService:
         Status transitions that trigger inventory operations:
         - Any → processing: Reserve inventory (if not already reserved)
         - processing → shipped: Fulfill inventory (deduct from stock)
+        - processing → partially_shipped: Partial fulfillment already done via fulfill_item
         - Any → cancelled: Release reservations
 
         Args:
@@ -491,11 +660,18 @@ class OrderInventoryService:
             result = self.reserve_for_order(db, vendor_id, order_id, skip_missing=True)
             logger.info(f"Order {order_id} confirmed: inventory reserved")
 
-        # Transitioning to shipped - fulfill inventory
+        # Transitioning to shipped - fulfill remaining inventory
         elif new_status == "shipped":
             result = self.fulfill_order(db, vendor_id, order_id, skip_missing=True)
             logger.info(f"Order {order_id} shipped: inventory fulfilled")
 
+        # partially_shipped - no automatic fulfillment (handled via fulfill_item)
+        elif new_status == "partially_shipped":
+            logger.info(
+                f"Order {order_id} partially shipped: use fulfill_item for item-level fulfillment"
+            )
+            result = {"order_id": order_id, "status": "partially_shipped"}
+
         # Transitioning to cancelled - release reservations
         elif new_status == "cancelled":
             # Only release if there was a previous status (order was in progress)
@@ -507,6 +683,53 @@ class OrderInventoryService:
 
         return result
 
+    def get_shipment_status(
+        self,
+        db: Session,
+        vendor_id: int,
+        order_id: int,
+    ) -> dict:
+        """
+        Get detailed shipment status for an order.
+
+        Returns item-level shipment status for partial shipment tracking.
+
+        Args:
+            db: Database session
+            vendor_id: Vendor ID
+            order_id: Order ID
+
+        Returns:
+            Dict with shipment status details
+        """
+        order = self.get_order_with_items(db, vendor_id, order_id)
+
+        items = []
+        for item in order.items:
+            items.append({
+                "item_id": item.id,
+                "product_id": item.product_id,
+                "product_name": item.product_name,
+                "quantity": item.quantity,
+                "shipped_quantity": item.shipped_quantity,
+                "remaining_quantity": item.remaining_quantity,
+                "is_fully_shipped": item.is_fully_shipped,
+                "is_partially_shipped": item.is_partially_shipped,
+            })
+
+        return {
+            "order_id": order_id,
+            "order_number": order.order_number,
+            "order_status": order.status,
+            "is_fully_shipped": order.is_fully_shipped,
+            "is_partially_shipped": order.is_partially_shipped,
+            "shipped_item_count": order.shipped_item_count,
+            "total_item_count": len(order.items),
+            "total_shipped_units": order.total_shipped_units,
+            "total_ordered_units": order.total_ordered_units,
+            "items": items,
+        }
+
 
 # Create service instance
 order_inventory_service = OrderInventoryService()

app/services/order_service.py

@@ -925,6 +925,7 @@ class OrderService:
         stats = {
             "pending": 0,
             "processing": 0,
+            "partially_shipped": 0,
             "shipped": 0,
             "delivered": 0,
             "cancelled": 0,
@@ -989,6 +990,9 @@ class OrderService:
             # Update timestamps based on status
             if order_update.status == "processing" and not order.confirmed_at:
                 order.confirmed_at = now
+            elif order_update.status == "partially_shipped":
+                # partially_shipped doesn't set shipped_at yet
+                pass
             elif order_update.status == "shipped" and not order.shipped_at:
                 order.shipped_at = now
             elif order_update.status == "delivered" and not order.delivered_at:
@@ -1256,6 +1260,7 @@ class OrderService:
             "total_orders": 0,
             "pending_orders": 0,
             "processing_orders": 0,
+            "partially_shipped_orders": 0,
             "shipped_orders": 0,
             "delivered_orders": 0,
             "cancelled_orders": 0,

docs/implementation/stock-management-integration.md

@@ -242,9 +242,130 @@ recent = db.query(InventoryTransaction).filter(
 ).order_by(InventoryTransaction.created_at.desc()).limit(10).all()
 ```
 
+## Partial Shipments (Phase 3)
+
+Orders can be partially shipped, allowing vendors to ship items as they become available.
+
+### Status Flow
+
+```
+pending → processing → partially_shipped → shipped → delivered
+                    ↘                     ↗
+                      → shipped (if all items shipped at once)
+```
+
+### OrderItem Tracking
+
+Each order item has a `shipped_quantity` field:
+
+```python
+class OrderItem:
+    quantity: int           # Total ordered
+    shipped_quantity: int   # Units shipped so far
+
+    @property
+    def remaining_quantity(self):
+        return self.quantity - self.shipped_quantity
+
+    @property
+    def is_fully_shipped(self):
+        return self.shipped_quantity >= self.quantity
+```
+
+### API Endpoints
+
+#### Get Shipment Status
+
+```http
+GET /api/v1/vendor/orders/{order_id}/shipment-status
+```
+
+Returns item-level shipment status:
+```json
+{
+    "order_id": 123,
+    "order_number": "ORD-1-20260101-ABC123",
+    "order_status": "partially_shipped",
+    "is_fully_shipped": false,
+    "is_partially_shipped": true,
+    "shipped_item_count": 1,
+    "total_item_count": 3,
+    "total_shipped_units": 2,
+    "total_ordered_units": 5,
+    "items": [
+        {
+            "item_id": 1,
+            "product_name": "Widget A",
+            "quantity": 2,
+            "shipped_quantity": 2,
+            "remaining_quantity": 0,
+            "is_fully_shipped": true
+        },
+        {
+            "item_id": 2,
+            "product_name": "Widget B",
+            "quantity": 3,
+            "shipped_quantity": 0,
+            "remaining_quantity": 3,
+            "is_fully_shipped": false
+        }
+    ]
+}
+```
+
+#### Ship Individual Item
+
+```http
+POST /api/v1/vendor/orders/{order_id}/items/{item_id}/ship
+Content-Type: application/json
+
+{
+    "quantity": 2  // Optional - defaults to remaining quantity
+}
+```
+
+Response:
+```json
+{
+    "order_id": 123,
+    "item_id": 1,
+    "fulfilled_quantity": 2,
+    "shipped_quantity": 2,
+    "remaining_quantity": 0,
+    "is_fully_shipped": true
+}
+```
+
+### Automatic Status Updates
+
+When shipping items:
+1. If some items are shipped → status becomes `partially_shipped`
+2. If all items are fully shipped → status becomes `shipped`
+
+### Service Usage
+
+```python
+from app.services.order_inventory_service import order_inventory_service
+
+# Ship partial quantity of an item
+result = order_inventory_service.fulfill_item(
+    db=db,
+    vendor_id=vendor_id,
+    order_id=order_id,
+    item_id=item_id,
+    quantity=2,  # Ship 2 units
+)
+
+# Get shipment status
+status = order_inventory_service.get_shipment_status(
+    db=db,
+    vendor_id=vendor_id,
+    order_id=order_id,
+)
+```
+
 ## Future Enhancements
 
 1. **Multi-Location Selection** - Choose which location to draw from
 2. **Backorder Support** - Handle orders when stock is insufficient
 3. **Return Processing** - Increase stock when orders are returned
-4. **Transaction API** - Endpoint to view inventory history

models/database/order.py

@@ -239,6 +239,37 @@ class Order(Base, TimestampMixin):
         """Check if this is a marketplace order."""
         return self.channel != "direct"
 
+    @property
+    def is_fully_shipped(self) -> bool:
+        """Check if all items are fully shipped."""
+        if not self.items:
+            return False
+        return all(item.is_fully_shipped for item in self.items)
+
+    @property
+    def is_partially_shipped(self) -> bool:
+        """Check if some items are shipped but not all."""
+        if not self.items:
+            return False
+        has_shipped = any(item.shipped_quantity > 0 for item in self.items)
+        all_shipped = all(item.is_fully_shipped for item in self.items)
+        return has_shipped and not all_shipped
+
+    @property
+    def shipped_item_count(self) -> int:
+        """Count of fully shipped items."""
+        return sum(1 for item in self.items if item.is_fully_shipped)
+
+    @property
+    def total_shipped_units(self) -> int:
+        """Total quantity shipped across all items."""
+        return sum(item.shipped_quantity for item in self.items)
+
+    @property
+    def total_ordered_units(self) -> int:
+        """Total quantity ordered across all items."""
+        return sum(item.quantity for item in self.items)
+
 
 class OrderItem(Base, TimestampMixin):
     """
@@ -280,6 +311,9 @@ class OrderItem(Base, TimestampMixin):
     inventory_reserved = Column(Boolean, default=False)
     inventory_fulfilled = Column(Boolean, default=False)
 
+    # === Shipment Tracking ===
+    shipped_quantity = Column(Integer, default=0, nullable=False)  # Units shipped so far
+
     # === Exception Tracking ===
     # True if product was not found by GTIN during import (linked to placeholder)
     needs_product_match = Column(Boolean, default=False, index=True)
@@ -342,3 +376,20 @@ class OrderItem(Base, TimestampMixin):
         if not self.exception:
             return False
         return self.exception.blocks_confirmation
+
+    # === SHIPMENT PROPERTIES ===
+
+    @property
+    def remaining_quantity(self) -> int:
+        """Quantity not yet shipped."""
+        return max(0, self.quantity - self.shipped_quantity)
+
+    @property
+    def is_fully_shipped(self) -> bool:
+        """Check if all units have been shipped."""
+        return self.shipped_quantity >= self.quantity
+
+    @property
+    def is_partially_shipped(self) -> bool:
+        """Check if some but not all units have been shipped."""
+        return 0 < self.shipped_quantity < self.quantity