📖 Looking for karrio's legacy docs? Visit docs.karrio.io

Data Mapping

The core of a Karrio carrier integration is the data mapping layer, which transforms between Karrio’s unified API format and carrier-specific formats. This section covers how to implement mapping functions for different carrier operations.

Mapper Implementation

The Mapper class is implemented in the karrio/mappers/[carrier_name]/mapper.py file. It extends the base karrio.api.mapper.Mapper class and provides implementations for various operations.

1import typing
2import karrio.lib as lib
3import karrio.api.mapper as mapper
4import karrio.core.models as models
5
6import karrio.mappers.[carrier_name].settings as provider_settings
7import karrio.providers.[carrier_name].rate as provider_rate
8import karrio.providers.[carrier_name].shipment as provider_shipment
9import karrio.providers.[carrier_name].tracking as provider_tracking
10import karrio.providers.[carrier_name].address as provider_address
11import karrio.providers.[carrier_name].utils as provider_utils
12
13
14class Mapper(mapper.Mapper):
15    """Mapper for Freight Express carrier."""
16
17    settings: provider_settings.Settings
18
19    def create_shipment(
20        self, payload: models.ShipmentRequest
21    ) -> typing.Tuple[dict, typing.List[models.Message]]:
22        """Create a shipment with Freight Express."""
23        request = provider_shipment.shipment_request(payload, self.settings)
24        response = self.proxy.create_shipment(request)
25        return provider_shipment.parse_shipment_response(response, self.settings)
26
27    def cancel_shipment(
28        self, payload: models.ShipmentCancelRequest
29    ) -> typing.Tuple[dict, typing.List[models.Message]]:
30        """Cancel a shipment with Freight Express."""
31        request = provider_shipment.cancel_shipment_request(payload, self.settings)
32        response = self.proxy.cancel_shipment(request)
33        return provider_shipment.parse_cancel_response(response, self.settings)
34
35    def get_rates(
36        self, payload: models.RateRequest
37    ) -> typing.Tuple[typing.List[models.RateDetails], typing.List[models.Message]]:
38        """Get shipping rates from Freight Express."""
39        request = provider_rate.rate_request(payload, self.settings)
40        response = self.proxy.get_rates(request)
41        return provider_rate.parse_rate_response(response, self.settings)
42
43    def get_tracking(
44        self, payload: models.TrackingRequest
45    ) -> typing.Tuple[typing.List[models.TrackingDetails], typing.List[models.Message]]:
46        """Track a shipment with Freight Express."""
47        request = provider_tracking.tracking_request(payload, self.settings)
48        response = self.proxy.get_tracking(request)
49        return provider_tracking.parse_tracking_response(response, self.settings)
50
51    def validate_address(
52        self, payload: models.AddressValidationRequest
53    ) -> typing.Tuple[typing.List[models.AddressValidationDetails], typing.List[models.Message]]:
54        """Validate an address with Freight Express."""
55        request = provider_address.address_validation_request(payload, self.settings)
56        response = self.proxy.validate_address(request)
57        return provider_address.parse_validation_response(response, self.settings)

The Mapper class delegates the actual mapping to provider-specific modules:

  • karrio/providers/[carrier_name]/rate.py - Rating operations
  • karrio/providers/[carrier_name]/shipment/ - Shipping operations
  • karrio/providers/[carrier_name]/tracking.py - Tracking operations
  • karrio/providers/[carrier_name]/address.py - Address validation operations

Provider Mapping Functions

For each operation, there are typically two mapping functions:

  1. Request Mapping: Transforms Karrio’s unified request into carrier-specific format
  2. Response Parsing: Transforms carrier-specific response into Karrio’s unified format

Request Mapping Example: Rating

Here’s an example of mapping a rate request:

1import typing
2import karrio.lib as lib
3import karrio.core.models as models
4import karrio.core.units as units
5import karrio.schemas.[carrier_name].rate_request as carrier
6import karrio.providers.[carrier_name].units as provider_units
7import karrio.providers.[carrier_name].utils as provider_utils
8
9
10def rate_request(
11    payload: models.RateRequest, settings: provider_utils.Settings
12) -> lib.Serializable:
13    """Create a rate request for Freight Express."""
14    # Extract data from the unified payload
15    shipper = lib.to_address(payload.shipper)
16    recipient = lib.to_address(payload.recipient)
17    packages = lib.to_packages(payload.parcels)
18    options = lib.to_shipping_options(payload.options)
19    services = lib.to_services(payload.services, provider_units.Service)
20
21    # Create the carrier-specific request
22    request = carrier.RateRequest(
23        shipFrom=carrier.Address(
24            name=shipper.person_name,
25            addressLine1=shipper.street,
26            addressLine2=shipper.address_line2,
27            city=shipper.city,
28            stateOrRegion=shipper.state_code,
29            postalCode=shipper.postal_code,
30            countryCode=shipper.country_code,
31            phoneNumber=shipper.phone_number,
32            email=shipper.email,
33        ),
34        shipTo=carrier.Address(
35            name=recipient.person_name,
36            addressLine1=recipient.street,
37            addressLine2=recipient.address_line2,
38            city=recipient.city,
39            stateOrRegion=recipient.state_code,
40            postalCode=recipient.postal_code,
41            countryCode=recipient.country_code,
42            phoneNumber=recipient.phone_number,
43            email=recipient.email,
44        ),
45        serviceTypes=list(services),
46        shipDate=lib.fdatetime(
47            options.shipment_date.state, "%Y-%m-%d", "%Y-%m-%dT%H:%M:%S.%fZ"
48        ),
49        containerSpecifications=[
50            carrier.ContainerSpecification(
51                dimensions=carrier.Dimensions(
52                    height=package.height.IN,
53                    length=package.length.IN,
54                    width=package.width.IN,
55                    unit="IN",
56                ),
57                weight=carrier.Weight(
58                    value=package.weight.LB,
59                    unit="LB",
60                ),
61            )
62            for package in packages
63        ],
64    )
65
66    # Return the request as a serializable object
67    return lib.Serializable(request, lib.to_dict)

Response Parsing Example: Rating

And here’s an example of parsing a rate response:

1import typing
2import karrio.lib as lib
3import karrio.core.models as models
4import karrio.core.errors as errors
5import karrio.schemas.[carrier_name].rate_response as carrier
6import karrio.providers.[carrier_name].error as provider_error
7import karrio.providers.[carrier_name].units as provider_units
8import karrio.providers.[carrier_name].utils as provider_utils
9
10
11def parse_rate_response(
12    _response: lib.Deserializable[dict],
13    settings: provider_utils.Settings,
14) -> typing.Tuple[typing.List[models.RateDetails], typing.List[models.Message]]:
15    """Parse a rate response from Freight Express."""
16    # Deserialize the raw response
17    response = _response.deserialize()
18
19    # Check for errors
20    errors: typing.List[models.Message] = sum(
21        [
22            provider_error.parse_error_response(data, settings)
23            for data in response.get("errors", [])
24        ],
25        [],
26    )
27
28    # Extract rate details
29    rates = [
30        _extract_rate_details(data, settings)
31        for data in response.get("serviceRates", [])
32    ]
33
34    return rates, errors
35
36
37def _extract_rate_details(
38    data: dict,
39    settings: provider_utils.Settings,
40) -> models.RateDetails:
41    """Extract rate details from a service rate."""
42    # Convert raw data to carrier-specific object
43    rate = lib.to_object(carrier.ServiceRate, data)
44
45    # Calculate transit days
46    transit = (
47        lib.to_date(rate.promise.deliveryWindow.start, "%Y-%m-%dT%H:%M:%S.%fZ").date()
48        - lib.to_date(rate.promise.receiveWindow.end, "%Y-%m-%dT%H:%M:%S.%fZ").date()
49    ).days
50
51    # Create a unified rate details object
52    return models.RateDetails(
53        carrier_id=settings.carrier_id,
54        carrier_name=settings.carrier_name,
55        service=provider_units.Service.map(rate.serviceType).name_or_key,
56        total_charge=lib.to_decimal(rate.totalCharge.value),
57        currency=rate.totalCharge.unit,
58        transit_days=transit,
59        extra_charges=[
60            models.ChargeDetails(
61                name=charge.name,
62                amount=lib.to_decimal(charge.value),
63                currency=charge.unit,
64            )
65            for charge in rate.charges
66        ],
67        meta=dict(
68            service_name=rate.serviceType,
69            service_code=rate.serviceCode,
70        ),
71    )

Common Mapping Patterns

Address Mapping

Converting between Karrio’s address format and carrier-specific formats:

1def address_request(
2    payload: models.AddressValidationRequest, settings: provider_utils.Settings
3) -> lib.Serializable:
4    """Create an address validation request."""
5    address = lib.to_address(payload.address)
6
7    request = carrier.AddressValidationRequest(
8        address=carrier.Address(
9            addressLine1=address.street,
10            addressLine2=address.address_line2,
11            city=address.city,
12            stateOrRegion=address.state_code,
13            postalCode=address.postal_code,
14            countryCode=address.country_code,
15        ),
16    )
17
18    return lib.Serializable(request, lib.to_dict)

Shipment Mapping

Creating a shipment request, which is often more complex:

1def shipment_request(
2    payload: models.ShipmentRequest, settings: provider_utils.Settings
3) -> lib.Serializable:
4    """Create a shipment request."""
5    shipper = lib.to_address(payload.shipper)
6    recipient = lib.to_address(payload.recipient)
7    packages = lib.to_packages(payload.parcels)
8    options = lib.to_shipping_options(payload.options)
9    services = lib.to_services(payload.service, provider_units.Service)
10    service = next(iter(services), None)
11    payment = payload.payment
12
13    # Handle customs for international shipments
14    customs = None
15    if payload.customs:
16        customs = carrier.Customs(
17            contents=payload.customs.content_type,
18            nonDelivery=payload.customs.non_delivery,
19            items=[
20                carrier.CustomsItem(
21                    description=item.description,
22                    originCountry=item.origin_country,
23                    quantity=item.quantity,
24                    weight=item.weight.value,
25                    value=item.value_amount,
26                    hsCode=item.hs_code,
27                )
28                for item in payload.customs.commodities
29            ],
30        )
31
32    # Create the shipment request
33    request = carrier.ShipmentRequest(
34        shipFrom=carrier.Address(
35            name=shipper.person_name,
36            addressLine1=shipper.street,
37            addressLine2=shipper.address_line2,
38            city=shipper.city,
39            stateOrRegion=shipper.state_code,
40            postalCode=shipper.postal_code,
41            countryCode=shipper.country_code,
42            phoneNumber=shipper.phone_number,
43            email=shipper.email,
44        ),
45        shipTo=carrier.Address(
46            name=recipient.person_name,
47            addressLine1=recipient.street,
48            addressLine2=recipient.address_line2,
49            city=recipient.city,
50            stateOrRegion=recipient.state_code,
51            postalCode=recipient.postal_code,
52            countryCode=recipient.country_code,
53            phoneNumber=recipient.phone_number,
54            email=recipient.email,
55        ),
56        serviceType=service,
57        shipDate=lib.fdatetime(
58            options.shipment_date.state, "%Y-%m-%d", "%Y-%m-%dT%H:%M:%S.%fZ"
59        ),
60        packageCount=len(packages),
61        returnService=options.return_service or False,
62        signatureRequired=options.signature_confirmation or False,
63        packages=[
64            carrier.Package(
65                dimensions=carrier.Dimensions(
66                    height=package.height.IN,
67                    length=package.length.IN,
68                    width=package.width.IN,
69                    unit="IN",
70                ),
71                weight=carrier.Weight(
72                    value=package.weight.LB,
73                    unit="LB",
74                ),
75                reference=package.reference or "",
76            )
77            for package in packages
78        ],
79        billing=carrier.Billing(
80            accountNumber=settings.account_number,
81            paymentType=payment.paid_by,
82        ),
83        customs=customs,
84        reference=payload.reference,
85        labelFormat=options.label_format or "PDF",
86    )
87
88    return lib.Serializable(request, lib.to_dict)

Tracking Mapping

Creating a tracking request, which is often simpler:

1def tracking_request(
2    payload: models.TrackingRequest, settings: provider_utils.Settings
3) -> lib.Serializable:
4    """Create a tracking request."""
5    # Tracking is often just a list of tracking numbers
6    request = {
7        "tracking_ids": payload.tracking_numbers,
8    }
9
10    return lib.Serializable(request, dict)

Unified Data Model

Karrio’s unified data model is defined in the karrio.core.models module. It provides standard classes for various shipping operations:

  • AddressValidationRequest / AddressValidationDetails
  • RateRequest / RateDetails
  • ShipmentRequest / ShipmentDetails
  • TrackingRequest / TrackingDetails
  • PickupRequest / PickupDetails

Your mapping functions should transform between these unified models and carrier-specific formats.

Helper Functions

Karrio provides several helper functions to simplify common mapping tasks:

Convert a payload address to a standardized Address object
1address = lib.to_address(payload.address)
2
3# Convert a list of parcels to standardized Package objects
4packages = lib.to_packages(payload.parcels)
5
6# Convert shipping options to a standardized ShippingOptions object
7options = lib.to_shipping_options(payload.options)
8
9# Convert service string(s) to carrier-specific service codes
10services = lib.to_services(payload.services, provider_units.Service)
11
12# Format a date or datetime string
13formatted_date = lib.fdatetime(
14    date_string, "%Y-%m-%d", "%Y-%m-%dT%H:%M:%S.%fZ"
15)
16
17# Convert a string to a decimal value
18amount = lib.to_decimal("123.45")
19
20# Convert a string to a date object
21date = lib.to_date("2023-09-15", "%Y-%m-%d")
22
23# Convert a dict to a specific object type
24obj = lib.to_object(carrier.Address, address_dict)

Best Practices

When implementing data mapping:

  1. Stay Pure: Keep mapping functions pure (no side effects)
  2. Handle Missing Data: Provide defaults for optional fields
  3. Validate Data: Check for required fields and valid formats
  4. Type Safety: Use type hints to ensure type safety
  5. Error Handling: Gracefully handle mapping errors
  6. Documentation: Document complex mapping logic
  7. Unit Testing: Write tests for mapping functions

In the next section, we’ll cover how to test your carrier integration.