Data Mapping

The core of a Karrio carrier integration is the data mapping layer. This layer transforms data between Karrio’s unified models and the specific formats required by the carrier’s API.

This transformation logic lives in two places:

The Mapper class in karrio/mappers/[carrier_name]/mapper.py
The provider functions in karrio/providers/[carrier_name]/

Mapper Implementation

The Mapper class connects Karrio’s core logic to your provider-specific implementation. It inherits from karrio.api.mapper.Mapper and delegates to functions in your provider module.

karrio/mappers/[carrier_name]/mapper.py
1import typing
2import karrio.lib as lib
3import karrio.api.mapper as mapper
4import karrio.core.models as models
5import karrio.providers.[carrier_name] as provider
6import karrio.mappers.[carrier_name].settings as provider_settings
7
8class Mapper(mapper.Mapper):
9    settings: provider_settings.Settings
10
11    # Request Creation Methods
12    def create_rate_request(
13        self, payload: models.RateRequest
14    ) -> lib.Serializable:
15        return provider.rate_request(payload, self.settings)
16
17    def create_shipment_request(
18        self, payload: models.ShipmentRequest
19    ) -> lib.Serializable:
20        return provider.shipment_request(payload, self.settings)
21
22    def create_tracking_request(
23        self, payload: models.TrackingRequest
24    ) -> lib.Serializable:
25        return provider.tracking_request(payload, self.settings)
26
27    # Response Parsing Methods
28    def parse_rate_response(
29        self, response: lib.Deserializable[str]
30    ) -> typing.Tuple[typing.List[models.RateDetails], typing.List[models.Message]]:
31        return provider.parse_rate_response(response, self.settings)
32
33    def parse_shipment_response(
34        self, response: lib.Deserializable[str]
35    ) -> typing.Tuple[models.ShipmentDetails, typing.List[models.Message]]:
36        return provider.parse_shipment_response(response, self.settings)
37
38    def parse_tracking_response(
39        self, response: lib.Deserializable[str]
40    ) -> typing.Tuple[typing.List[models.TrackingDetails], typing.List[models.Message]]:
41        return provider.parse_tracking_response(response, self.settings)

Provider Mapping Functions

The actual mapping logic resides in functions within your provider modules. For each operation, you implement two functions: one to create the request and one to parse the response.

Request Mapping Example: Rating

File: karrio/providers/[carrier_name]/rate.py

1import typing
2import karrio.lib as lib
3import karrio.core.models as models
4import karrio.providers.[carrier_name].units as provider_units
5import karrio.providers.[carrier_name].utils as provider_utils
6import karrio.providers.[carrier_name].error as provider_error
7# CRITICAL: Always import and use generated schema types
8import karrio.schemas.[carrier_name].rate_request as carrier_req
9import karrio.schemas.[carrier_name].rate_response as carrier_res
10
11def rate_request(
12    payload: models.RateRequest,
13    settings: provider_utils.Settings,
14) -> lib.Serializable:
15    """Transform a Karrio RateRequest into a carrier-specific rate request."""
16
17    # Use Karrio helpers to parse the unified payload
18    shipper = lib.to_address(payload.shipper)
19    recipient = lib.to_address(payload.recipient)
20    packages = lib.to_packages(payload.parcels)
21    services = lib.to_services(payload.services, provider_units.ShippingService)
22    options = lib.to_shipping_options(
23        payload.options,
24        package_options=packages.options,
25        initializer=provider_units.shipping_options_initializer,
26    )
27
28    # Create the carrier-specific request using generated schema types
29    request = carrier_req.RateRequestType(
30        shipper={
31            "addressLine1": shipper.address_line1,
32            "city": shipper.city,
33            "postalCode": shipper.postal_code,
34            "countryCode": shipper.country_code,
35            "stateCode": shipper.state_code,
36            "personName": shipper.person_name,
37            "companyName": shipper.company_name,
38        },
39        recipient={
40            "addressLine1": recipient.address_line1,
41            "city": recipient.city,
42            "postalCode": recipient.postal_code,
43            "countryCode": recipient.country_code,
44            "stateCode": recipient.state_code,
45            "personName": recipient.person_name,
46            "companyName": recipient.company_name,
47        },
48        packages=[
49            {
50                "weight": package.weight.value,
51                "weightUnit": provider_units.WeightUnit[package.weight.unit].value,
52                "length": package.length.value if package.length else None,
53                "width": package.width.value if package.width else None,
54                "height": package.height.value if package.height else None,
55                "dimensionUnit": provider_units.DimensionUnit[package.dimension_unit].value if package.dimension_unit else None,
56                "packagingType": provider_units.PackagingType[package.packaging_type or 'your_packaging'].value,
57            }
58            for package in packages
59        ],
60        services=[s.value_or_key for s in services] if services else None,
61        customerNumber=settings.account_number,
62    )
63
64    return lib.Serializable(request, lib.to_dict)  # Use lib.to_xml for XML APIs
65
66def parse_rate_response(
67    _response: lib.Deserializable[str],
68    settings: provider_utils.Settings,
69) -> typing.Tuple[typing.List[models.RateDetails], typing.List[models.Message]]:
70    """Parse a carrier rate response into Karrio RateDetails."""
71    response = _response.deserialize()
72
73    # Parse any errors first
74    messages = provider_error.parse_error_response(response, settings)
75
76    # Extract rates from response
77    rate_objects = response.get("rates", []) if hasattr(response, 'get') else []
78    rates = [_extract_rate_details(rate, settings) for rate in rate_objects]
79
80    return rates, messages
81
82def _extract_rate_details(
83    data: dict,
84    settings: provider_utils.Settings,
85) -> models.RateDetails:
86    """Helper function to map a single carrier rate to RateDetails."""
87    # Convert to typed object using generated schema
88    rate = lib.to_object(carrier_res.RateType, data)
89
90    return models.RateDetails(
91        carrier_name=settings.carrier_name,
92        carrier_id=settings.carrier_id,
93        service=rate.serviceCode if hasattr(rate, 'serviceCode') else "",
94        total_charge=lib.to_money(rate.totalCharge),
95        currency=rate.currency if hasattr(rate, 'currency') else "USD",
96        transit_days=int(rate.transitDays) if hasattr(rate, 'transitDays') and rate.transitDays else None,
97        meta=dict(
98            service_name=rate.serviceName if hasattr(rate, 'serviceName') else "",
99        ),
100    )

Shipment Implementation Example

File: karrio/providers/[carrier_name]/shipment.py

1import typing
2import karrio.lib as lib
3import karrio.core.models as models
4import karrio.providers.[carrier_name].units as provider_units
5import karrio.providers.[carrier_name].utils as provider_utils
6import karrio.providers.[carrier_name].error as provider_error
7import karrio.schemas.[carrier_name].shipment_request as carrier_req
8import karrio.schemas.[carrier_name].shipment_response as carrier_res
9
10def shipment_request(
11    payload: models.ShipmentRequest,
12    settings: provider_utils.Settings,
13) -> lib.Serializable:
14    """Create a shipment request for the carrier API."""
15
16    shipper = lib.to_address(payload.shipper)
17    recipient = lib.to_address(payload.recipient)
18    packages = lib.to_packages(payload.parcels)
19    service = lib.to_services(payload.service, provider_units.ShippingService).first
20    options = lib.to_shipping_options(
21        payload.options,
22        package_options=packages.options,
23        initializer=provider_units.shipping_options_initializer,
24    )
25
26    request = carrier_req.ShipmentRequestType(
27        shipper={
28            "addressLine1": shipper.address_line1,
29            "city": shipper.city,
30            "postalCode": shipper.postal_code,
31            "countryCode": shipper.country_code,
32            "personName": shipper.person_name,
33            "companyName": shipper.company_name,
34        },
35        recipient={
36            "addressLine1": recipient.address_line1,
37            "city": recipient.city,
38            "postalCode": recipient.postal_code,
39            "countryCode": recipient.country_code,
40            "personName": recipient.person_name,
41            "companyName": recipient.company_name,
42        },
43        packages=[
44            {
45                "weight": package.weight.value,
46                "weightUnit": provider_units.WeightUnit[package.weight.unit].value,
47                "dimensions": {
48                    "length": package.length.value if package.length else None,
49                    "width": package.width.value if package.width else None,
50                    "height": package.height.value if package.height else None,
51                    "unit": provider_units.DimensionUnit[package.dimension_unit].value if package.dimension_unit else None,
52                },
53                "packagingType": provider_units.PackagingType[package.packaging_type or 'your_packaging'].value,
54            }
55            for package in packages
56        ],
57        serviceCode=service.value_or_key,
58        customerNumber=settings.account_number,
59        labelFormat=payload.label_type or "PDF",
60    )
61
62    return lib.Serializable(request, lib.to_dict)
63
64def parse_shipment_response(
65    _response: lib.Deserializable[str],
66    settings: provider_utils.Settings,
67) -> typing.Tuple[models.ShipmentDetails, typing.List[models.Message]]:
68    """Parse a carrier shipment response into ShipmentDetails."""
69    response = _response.deserialize()
70
71    messages = provider_error.parse_error_response(response, settings)
72
73    # Convert to typed object
74    shipment = lib.to_object(carrier_res.ShipmentResponseType, response)
75
76    return models.ShipmentDetails(
77        carrier_id=settings.carrier_id,
78        carrier_name=settings.carrier_name,
79        tracking_number=shipment.trackingNumber,
80        shipment_identifier=shipment.shipmentId if hasattr(shipment, 'shipmentId') else None,
81        label=shipment.labelData if hasattr(shipment, 'labelData') else None,
82        meta=dict(
83            service_name=shipment.serviceName if hasattr(shipment, 'serviceName') else "",
84            label_type=shipment.labelType if hasattr(shipment, 'labelType') else "PDF",
85        ),
86    ), messages

Tracking Implementation Example

File: karrio/providers/[carrier_name]/tracking.py

1import typing
2import karrio.lib as lib
3import karrio.core.models as models
4import karrio.providers.[carrier_name].units as provider_units
5import karrio.providers.[carrier_name].utils as provider_utils
6import karrio.schemas.[carrier_name].tracking_response as carrier_res
7
8def tracking_request(
9    payload: models.TrackingRequest,
10    settings: provider_utils.Settings,
11) -> lib.Serializable:
12    """Create a tracking request object."""
13    # Most carriers just need the tracking numbers
14    return lib.Serializable(payload.tracking_numbers)
15
16def parse_tracking_response(
17    _response: lib.Deserializable,
18    settings: provider_utils.Settings,
19) -> typing.Tuple[typing.List[models.TrackingDetails], typing.List[models.Message]]:
20    """Parse carrier tracking response into TrackingDetails."""
21    response = _response.deserialize()
22
23    tracking_details = []
24    for tracking_number, tracking_data in response:
25        if not tracking_data:
26            continue
27
28        # Convert to typed object
29        tracking = lib.to_object(carrier_res.TrackingResponseType, tracking_data)
30
31        events = [
32            models.TrackingEvent(
33                date=lib.to_date(event.date) if hasattr(event, 'date') else None,
34                description=event.description if hasattr(event, 'description') else "",
35                location=event.location if hasattr(event, 'location') else "",
36                code=event.status if hasattr(event, 'status') else "",
37                time=lib.to_time(event.time) if hasattr(event, 'time') else None,
38            )
39            for event in (getattr(tracking, 'events', None) or [])
40        ]
41
42        detail = models.TrackingDetails(
43            carrier_id=settings.carrier_id,
44            carrier_name=settings.carrier_name,
45            tracking_number=tracking_number,
46            events=events,
47            status=provider_units.TrackingStatus.map(getattr(tracking, 'status', None)),
48        )
49        tracking_details.append(detail)
50
51    return tracking_details, []

Helper Functions

Karrio provides helper functions in karrio.lib to simplify common mapping tasks:

lib.to_address: Converts payload address to standardized Address object
lib.to_packages: Converts parcels list to standardized Package objects
lib.to_shipping_options: Converts options to ShippingOptions object
lib.to_services: Converts service strings to carrier-specific service codes
lib.to_money: Converts values to Decimal money type
lib.to_object: Converts dictionaries to specific schema class objects
lib.to_date: Parses date strings into date objects
lib.to_time: Parses time strings into time objects

Unified Data Models

Karrio’s unified data model is defined in karrio.core.models:

Request Models

RateRequest: Request for shipping rates
ShipmentRequest: Request to create shipment
TrackingRequest: Request to track shipments
PickupRequest: Request to schedule pickup

Response Models

RateDetails: Individual rate quote
ShipmentDetails: Created shipment information
TrackingDetails: Tracking status and events
PickupDetails: Pickup confirmation

Supporting Models

Address: Standardized address format
Package: Package dimensions and weight
Message: Error or warning message

Best Practices

Always Use Generated Types: Import and use schema classes from karrio.schemas.[carrier_name]
Handle Missing Data: Provide sensible defaults for optional fields
Use Helper Functions: Leverage karrio.lib helpers for data conversion
Map to Unified Types: Always map from carrier-specific to unified formats
Error Handling: Parse and return carrier errors as Message objects
Type Safety: Use type hints and proper typing throughout
Stay Pure: Mapping functions should be side-effect free

Common Patterns

Address Mapping

1def _build_address(address: models.Address) -> dict:
2    return {
3        "addressLine1": address.address_line1,
4        "city": address.city,
5        "postalCode": address.postal_code,
6        "countryCode": address.country_code,
7        "stateCode": address.state_code,
8        "personName": address.person_name,
9        "companyName": address.company_name,
10    }

Service Code Mapping

1def _map_service_code(service_code: str) -> str:
2    service_map = {
3        "CARRIER_EXPRESS": "carrier_express",
4        "CARRIER_STANDARD": "carrier_standard",
5        "CARRIER_GROUND": "carrier_ground",
6    }
7    return service_map.get(service_code, service_code.lower())

Weight/Dimension Conversion

1def _build_weight(package: models.Package) -> dict:
2    return {
3        "value": package.weight.value,
4        "unit": "KG" if package.weight.unit == "KG" else "LB"
5    }
6
7def _build_dimensions(package: models.Package) -> dict:
8    if not package.length:
9        return None
10    return {
11        "length": package.length.value,
12        "width": package.width.value,
13        "height": package.height.value,
14        "unit": "CM" if package.dimension_unit == "CM" else "IN"
15    }

Error Handling Pattern

1def parse_error_response(
2    response: dict,
3    settings: provider_utils.Settings,
4) -> typing.List[models.Message]:
5    """Parse carrier errors into Karrio Messages."""
6    errors = response.get("errors", [])
7
8    return [
9        models.Message(
10            carrier_id=settings.carrier_id,
11            carrier_name=settings.carrier_name,
12            code=error.get("code", ""),
13            message=error.get("message", ""),
14            details=error.get("details", {}),
15        )
16        for error in errors
17    ]

Integration Flow

Testing Data Mapping

CRITICAL: Every mapping function must be tested with the exact patterns:

1. Request Transformation Test

1def test_create_rate_request(self):
2    request = gateway.mapper.create_rate_request(self.RateRequest)
3    print(f"Generated request: {lib.to_dict(request.serialize())}")
4    self.assertEqual(lib.to_dict(request.serialize()), RateRequest)

2. Response Parsing Test

1def test_parse_rate_response(self):
2    with patch("karrio.mappers.[carrier_id].proxy.lib.request") as mock:
3        mock.return_value = RateResponse
4        parsed_response = (
5            karrio.Rating.fetch(self.RateRequest)
6            .from_(gateway)
7            .parse()
8        )
9        print(f"Parsed response: {lib.to_dict(parsed_response)}")
10        self.assertListEqual(lib.to_dict(parsed_response), ParsedRateResponse)

3. Test Data Requirements

Input Payload: Standard Karrio format with realistic test data
Expected Request: Carrier-specific format that your rate_request() should produce
Mock Response: Actual carrier API response format
Expected Output: Karrio format with [data, errors] structure

4. Generated Types Validation

Always test that your mapping correctly uses generated schema types:

Verify request uses generated types
1rate = lib.to_object(carrier_req.RateRequestType, request.serialize())
2self.assertIsInstance(rate, carrier_req.RateRequestType)
3
4# Verify response parsing uses generated types
5response_obj = lib.to_object(carrier_res.RateResponseType, mock_data)
6self.assertTrue(hasattr(response_obj, 'rates'))

The data mapping layer is the heart of your integration, transforming between Karrio’s unified format and carrier-specific requirements while maintaining type safety and error handling throughout the process.

Getting Started

OSS contribution

Guides