API Requests

The Proxy class is responsible for handling communication with the carrier’s API. It defines methods for each API operation (rating, shipping, tracking, etc.) and handles authentication, request formatting, and response parsing.

Proxy Class Structure

The Proxy class is implemented in the karrio/mappers/[carrier_name]/proxy.py file and extends the base karrio.api.proxy.Proxy class:

1import karrio.lib as lib
2import karrio.api.proxy as proxy
3import karrio.mappers.[carrier_name].settings as provider_settings
4import karrio.providers.[carrier_name].utils as provider_utils
5
6
7class Proxy(proxy.Proxy):
8    """API client for the Freight Express carrier."""
9
10    settings: provider_settings.Settings
11
12    def get_rates(self, request: lib.Serializable) -> lib.Deserializable[dict]:
13        """Request shipping rates from the Freight Express API."""
14        endpoint = provider_utils.get_endpoint_url(self.settings, "shipping/v1/rates")
15        headers = provider_utils.get_auth_header(self.settings)
16
17        response = lib.request(
18            url=endpoint,
19            data=lib.to_json(request.serialize()),
20            headers={
21                **headers,
22                "Accept": "application/json",
23                "Content-Type": "application/json",
24            },
25            method="POST",
26            trace=self.trace_as("json"),
27        )
28
29        return lib.Deserializable(response, lib.to_dict)
30
31    def create_shipment(self, request: lib.Serializable) -> lib.Deserializable[dict]:
32        """Create a shipment and purchase a label."""
33        endpoint = provider_utils.get_endpoint_url(self.settings, "shipping/v1/shipments")
34        headers = provider_utils.get_auth_header(self.settings)
35
36        response = lib.request(
37            url=endpoint,
38            data=lib.to_json(request.serialize()),
39            headers={
40                **headers,
41                "Accept": "application/json",
42                "Content-Type": "application/json",
43            },
44            method="POST",
45            trace=self.trace_as("json"),
46        )
47
48        return lib.Deserializable(response, lib.to_dict)
49
50    def get_tracking(self, request: lib.Serializable) -> lib.Deserializable[dict]:
51        """Track a shipment by tracking number."""
52        data = request.serialize()
53        tracking_numbers = ",".join(data.get("tracking_ids", []))
54        endpoint = provider_utils.get_endpoint_url(
55            self.settings, f"tracking/v1/shipments?tracking_numbers={tracking_numbers}"
56        )
57        headers = provider_utils.get_auth_header(self.settings)
58
59        response = lib.request(
60            url=endpoint,
61            headers={
62                **headers,
63                "Accept": "application/json",
64            },
65            method="GET",
66            trace=self.trace_as("json"),
67        )
68
69        return lib.Deserializable(response, lib.to_dict)
70
71    # More API methods as needed...

HTTP Requests

The Proxy class uses the lib.request function to make HTTP requests to the carrier API. This function handles:

HTTP method (GET, POST, etc.)
Request headers
Request body
Response parsing
Error handling

For each API operation, you’ll typically:

Construct the API endpoint URL
Set up authentication headers
Serialize the request data (for POST/PUT requests)
Make the HTTP request
Return a Deserializable object with the response

Example: JSON API Request

For JSON-based APIs, requests typically look like this:

1def create_shipment(self, request: lib.Serializable) -> lib.Deserializable[dict]:
2    """Create a shipment and purchase a label."""
3    endpoint = f"{self.settings.server_url}/shipping/v1/shipments"
4    headers = {
5        "Authorization": f"Bearer {self.settings.api_key}",
6        "Accept": "application/json",
7        "Content-Type": "application/json",
8    }
9
10    response = lib.request(
11        url=endpoint,
12        data=lib.to_json(request.serialize()),
13        headers=headers,
14        method="POST",
15        trace=self.trace_as("json"),
16    )
17
18    return lib.Deserializable(response, lib.to_dict)

Example: XML API Request

For XML/SOAP-based APIs, requests look slightly different:

1def create_shipment(self, request: lib.Serializable) -> lib.Deserializable[str]:
2    """Create a shipment and purchase a label."""
3    endpoint = f"{self.settings.server_url}/soap/shipping"
4    headers = {
5        "Content-Type": "text/xml; charset=utf-8",
6        "SOAPAction": "http://api.carrier.com/shipping/create",
7    }
8
9    # For SOAP APIs, we often need to wrap the XML in a SOAP envelope
10    xml_request = f"""
11    <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
12                     xmlns:ship="http://api.carrier.com/shipping">
13        <soapenv:Header>
14            <ship:Authentication>
15                <ship:Username>{self.settings.username}</ship:Username>
16                <ship:Password>{self.settings.password}</ship:Password>
17            </ship:Authentication>
18        </soapenv:Header>
19        <soapenv:Body>
20            {request.serialize()}
21        </soapenv:Body>
22    </soapenv:Envelope>
23    """
24
25    response = lib.request(
26        url=endpoint,
27        data=xml_request,
28        headers=headers,
29        method="POST",
30        trace=self.trace_as("xml"),
31    )
32
33    return lib.Deserializable(response, lib.xml_to_dict)

Tracing and Debugging

Karrio includes a built-in tracing mechanism to help debug API calls. The trace_as method on the Proxy class sets up tracing for a particular format (JSON, XML, etc.):

1trace=self.trace_as("json")  # For JSON APIs
2trace=self.trace_as("xml")   # For XML APIs

This allows Karrio to record and display API requests and responses for debugging purposes.

Authentication

Authentication is typically handled by utility functions in the karrio/providers/[carrier_name]/utils.py file. These functions generate the appropriate headers based on the carrier’s authentication requirements:

1def get_auth_header(settings: Settings) -> Dict[str, str]:
2    """Generate authentication headers for API requests."""
3    if settings.api_key:
4        return {"Authorization": f"Bearer {settings.api_key}"}
5
6    # Basic auth
7    credentials = f"{settings.username}:{settings.password}"
8    encoded = base64.b64encode(credentials.encode()).decode()
9    return {"Authorization": f"Basic {encoded}"}

Error Handling

API errors should be handled gracefully by checking the response status code and content. Common error handling patterns include:

1response = lib.request(
2    url=endpoint,
3    data=lib.to_json(request.serialize()),
4    headers=headers,
5    method="POST",
6    trace=self.trace_as("json"),
7)
8
9# Check for API errors in the response
10if response.get("error"):
11    error_message = response.get("error", {}).get("message", "Unknown error")
12    # Handle error appropriately
13    # This will typically be handled by the error parsing function

A separate error parsing module (karrio/providers/[carrier_name]/error.py) is often used to standardize error handling:

1def parse_error_response(response: dict, settings: Settings) -> List[Message]:
2    """Parse error response into standardized Karrio Messages."""
3    errors = []
4
5    if "error" in response:
6        error = response["error"]
7        errors.append(
8            Message(
9                code=error.get("code", "ERROR"),
10                message=error.get("message", "Unknown error"),
11                carrier_name=settings.carrier_name,
12                carrier_id=settings.carrier_id,
13            )
14        )
15
16    # ... more error parsing logic
17
18    return errors

API Request Flow

The typical flow of an API request in Karrio is:

The user makes a request through Karrio’s unified API
The request is routed to the appropriate carrier extension
The Mapper transforms the unified request into a carrier-specific format
The Proxy makes the API call to the carrier
The carrier responds with data
The Mapper transforms the carrier-specific response back into Karrio’s unified format
The unified response is returned to the user

Best Practices

When implementing API requests:

Use Helper Functions: Extract common logic into utility functions
Handle Errors Gracefully: Always check for and handle API errors
Log Important Information: Use tracing to record API requests and responses
Respect Rate Limits: Implement retry logic with backoff for rate-limited APIs
Timeouts: Set appropriate timeouts for API calls
Clean Code: Keep HTTP request logic separate from business logic

Common API Operations

Most carrier integrations implement some or all of these common operations:

Rating: Get shipping rates for a package
Shipping: Create shipments and purchase labels
Tracking: Track shipments by tracking number
Address Validation: Validate and normalize shipping addresses
Pickup: Schedule, cancel, or update pickups
Customs: Generate customs documentation for international shipments

In the next section, we’ll cover how to implement data mapping between Karrio’s unified format and carrier-specific formats.

Getting Started

OSS contribution

Guides