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

API Requests

The Proxy class is responsible for handling all communication with the carrier’s API. It contains methods for each supported operation (rating, shipping, tracking, etc.), and manages the specifics of authentication, request formatting, and response retrieval.

Proxy Class Structure

The Proxy class is implemented in karrio/mappers/[carrier_name]/proxy.py and extends the base karrio.api.proxy.Proxy class. Its primary role is to make HTTP requests and return a lib.Deserializable object containing the raw response from the carrier.

File: karrio/mappers/[carrier_name]/proxy.py

1import karrio.lib as lib
2import karrio.api.proxy as proxy
3import karrio.mappers.[carrier_name].settings as provider_settings
4
5class Proxy(proxy.Proxy):
6    settings: provider_settings.Settings
7
8    def get_rates(self, request: lib.Serializable) -> lib.Deserializable[str]:
9        """Request shipping rates from the carrier's API."""
10
11        response = lib.request(
12            url=f"{self.settings.server_url}/rates",
13            data=lib.to_json(request.serialize()),  # Use request.serialize() for XML
14            trace=self.trace_as("json"),             # Use "xml" for XML APIs
15            method="POST",
16            headers={
17                "Content-Type": "application/json",  # Use "text/xml" for XML APIs
18                "Authorization": f"Bearer {self.settings.api_key}"  # Or access_token
19            },
20        )
21
22        return lib.Deserializable(response, lib.to_dict)  # Use lib.to_element for XML
23
24    def create_shipment(self, request: lib.Serializable) -> lib.Deserializable[str]:
25        """Create a shipment and request a shipping label."""
26
27        response = lib.request(
28            url=f"{self.settings.server_url}/shipments",
29            data=lib.to_json(request.serialize()),
30            trace=self.trace_as("json"),
31            method="POST",
32            headers={
33                "Content-Type": "application/json",
34                "Authorization": f"Bearer {self.settings.api_key}"
35            },
36        )
37
38        return lib.Deserializable(response, lib.to_dict)
39
40    def get_tracking(self, request: lib.Serializable) -> lib.Deserializable:
41        """Retrieve tracking details for one or more shipments."""
42
43        def _get_tracking(tracking_number: str):
44            return tracking_number, lib.request(
45                url=f"{self.settings.server_url}/tracking/{tracking_number}",
46                trace=self.trace_as("json"),
47                method="GET",
48                headers={"Authorization": f"Bearer {self.settings.api_key}"},
49            )
50
51        # Use concurrent requests for multiple tracking numbers
52        responses = lib.run_concurently(_get_tracking, request.serialize())
53
54        return lib.Deserializable(
55            responses,
56            lambda res: [
57                (num, lib.to_dict(track)) for num, track in res if any(track.strip())
58            ],
59        )
60
61    def cancel_shipment(self, request: lib.Serializable) -> lib.Deserializable[str]:
62        """Cancel a shipment."""
63
64        response = lib.request(
65            url=f"{self.settings.server_url}/shipments/cancel",
66            data=lib.to_json(request.serialize()),
67            trace=self.trace_as("json"),
68            method="POST",
69            headers={
70                "Content-Type": "application/json",
71                "Authorization": f"Bearer {self.settings.api_key}"
72            },
73        )
74
75        return lib.Deserializable(response, lib.to_dict)

HTTP Requests with lib.request

The lib.request function is a wrapper around Python’s requests library that provides:

  • HTTP method handling (POST, GET, DELETE)
  • Request headers and body management
  • Tracing for debugging
  • Basic error handling for network issues

The result is passed into a lib.Deserializable object with a parsing function (lib.to_dict for JSON, lib.to_element for XML).

JSON API Example

1def get_rates(self, request: lib.Serializable) -> lib.Deserializable[str]:
2    response = lib.request(
3        url=f"{self.settings.server_url}/rates",
4        data=lib.to_json(request.serialize()),
5        trace=self.trace_as("json"),
6        method="POST",
7        headers={
8            "Content-Type": "application/json",
9            "Authorization": f"Bearer {self.settings.api_key}"
10        },
11    )
12    return lib.Deserializable(response, lib.to_dict)

XML/SOAP API Example

1def get_rates(self, request: lib.Serializable) -> lib.Deserializable[str]:
2    response = lib.request(
3        url=f"{self.settings.server_url}/rates",
4        data=request.serialize(),  # XML string from generated schema
5        trace=self.trace_as("xml"),
6        method="POST",
7        headers={
8            "Content-Type": "text/xml",
9            "SOAPAction": "getRates",
10            "Authorization": f"Basic {self.settings.authorization}"
11        },
12    )
13    return lib.Deserializable(response, lib.to_element)

Authentication Patterns

Authentication logic is defined in the Settings class in karrio/providers/[carrier_name]/utils.py.

API Key Authentication (Bearer Token)

1class Settings(core.Settings):
2    api_key: str
3
4    @property
5    def auth_headers(self):
6        return {"Authorization": f"Bearer {self.api_key}"}

Usage in Proxy:

1headers={
2    "Content-Type": "application/json",
3    "Authorization": f"Bearer {self.settings.api_key}"
4}

Basic Authentication

1class Settings(core.Settings):
2    username: str
3    password: str
4
5    @property
6    def authorization(self):
7        pair = f"{self.username}:{self.password}"
8        return base64.b64encode(pair.encode("utf-8")).decode("ascii")

Usage in Proxy:

1headers={
2    "Content-Type": "application/json",
3    "Authorization": f"Basic {self.settings.authorization}"
4}

OAuth 2.0 Authentication

1class Settings(core.Settings):
2    client_id: str
3    client_secret: str
4
5    @property
6    def access_token(self):
7        """Retrieve and cache OAuth access token."""
8        cache_key = f"{self.carrier_name}|{self.client_id}|{self.client_secret}"
9        now = datetime.datetime.now() + datetime.timedelta(minutes=30)
10
11        auth = self.connection_cache.get(cache_key) or {}
12        token = auth.get("access_token")
13        expiry = lib.to_date(auth.get("expiry"), current_format="%Y-%m-%d %H:%M:%S")
14
15        if token is not None and expiry is not None and expiry > now:
16            return token
17
18        # Token expired or doesn't exist, get new one
19        self.connection_cache.set(cache_key, lambda: login(self))
20        new_auth = self.connection_cache.get(cache_key)
21        return new_auth["access_token"]

Usage in Proxy:

1headers={
2    "Content-Type": "application/json",
3    "Authorization": f"Bearer {self.settings.access_token}"
4}

Tracing and Debugging

Karrio includes built-in tracing to help debug API calls:

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

This records and displays the full request and response for each API call, invaluable for debugging.

Error Handling

The Proxy class only handles HTTP communication. It does not parse response content for business logic errors.

Error parsing is handled in the provider’s error.py module:

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

1import typing
2import karrio.lib as lib
3import karrio.core.models as models
4import karrio.providers.[carrier_name].utils as provider_utils
5
6def parse_error_response(
7    response: dict,  # Use lib.Element for XML APIs
8    settings: provider_utils.Settings,
9    **kwargs,
10) -> typing.List[models.Message]:
11    """Parse carrier error response into Karrio Messages."""
12
13    # Extract errors based on carrier API structure
14    errors = response.get("errors", []) if hasattr(response, 'get') else []
15
16    # For XML APIs:
17    # errors = response.xpath("//error") if hasattr(response, 'xpath') else []
18
19    return [
20        models.Message(
21            carrier_id=settings.carrier_id,
22            carrier_name=settings.carrier_name,
23            code=error.get("code", ""),
24            message=error.get("message", ""),
25            details={**kwargs, "details": error.get("details", "")},
26        )
27        for error in errors
28    ]

Common API Patterns

Single Request APIs

1def get_rates(self, request: lib.Serializable) -> lib.Deserializable[str]:
2    response = lib.request(
3        url=f"{self.settings.server_url}/rates",
4        data=lib.to_json(request.serialize()),
5        trace=self.trace_as("json"),
6        method="POST",
7        headers=self.settings.auth_headers,
8    )
9    return lib.Deserializable(response, lib.to_dict)

Bulk Request APIs

1def get_tracking(self, request: lib.Serializable) -> lib.Deserializable:
2    response = lib.request(
3        url=f"{self.settings.server_url}/tracking/bulk",
4        data=lib.to_json({"tracking_numbers": request.serialize()}),
5        trace=self.trace_as("json"),
6        method="POST",
7        headers=self.settings.auth_headers,
8    )
9    return lib.Deserializable(response, lib.to_dict)

Concurrent Individual Requests

1def get_tracking(self, request: lib.Serializable) -> lib.Deserializable:
2    def _get_tracking(tracking_number: str):
3        return tracking_number, lib.request(
4            url=f"{self.settings.server_url}/tracking/{tracking_number}",
5            trace=self.trace_as("json"),
6            method="GET",
7            headers=self.settings.auth_headers,
8        )
9
10    responses = lib.run_concurently(_get_tracking, request.serialize())
11
12    return lib.Deserializable(
13        responses,
14        lambda res: [
15            (num, lib.to_dict(track)) for num, track in res if any(track.strip())
16        ],
17    )

File Upload APIs

1def upload_document(self, request: lib.Serializable) -> lib.Deserializable[str]:
2    data = request.serialize()
3
4    response = lib.request(
5        url=f"{self.settings.server_url}/documents",
6        files={"document": ("document.pdf", data["content"], "application/pdf")},
7        data={"metadata": lib.to_json(data["metadata"])},
8        trace=self.trace_as("json"),
9        method="POST",
10        headers={"Authorization": f"Bearer {self.settings.api_key}"},
11    )
12
13    return lib.Deserializable(response, lib.to_dict)

API Request Flow

Best Practices

  1. Keep it Simple: The Proxy should only handle HTTP communication
  2. Use Settings Properties: Access all configuration via self.settings
  3. Implement Stubs First: Use stubbed responses during development
  4. Handle Timeouts: Set appropriate timeout values for different operations
  5. Use Concurrent Requests: For operations involving multiple requests
  6. Proper Error Handling: Let provider layer handle business logic errors
  7. Trace Everything: Use trace_as() for debugging all requests

Testing Proxy Implementation

Test with stubbed responses during development
1def get_rates(self, request: lib.Serializable) -> lib.Deserializable[str]:
2    # DEVELOPMENT ONLY: Replace with actual API call
3    stubbed_response = {
4        "rates": [
5            {
6                "serviceCode": "standard",
7                "serviceName": "Standard Service",
8                "totalCharge": 10.50,
9                "currency": "USD"
10            }
11        ]
12    }
13
14    return lib.Deserializable(lib.to_json(stubbed_response), lib.to_dict)

Testing Proxy Implementation

CRITICAL: Every proxy method must be tested with exact patterns:

1. Endpoint Verification Test

1def test_get_rates(self):
2    """Test that proxy calls correct API endpoint."""
3    with patch("karrio.mappers.[carrier_id].proxy.lib.request") as mock:
4        mock.return_value = "{}"  # Use "<r></r>" for XML APIs
5        karrio.Rating.fetch(self.RateRequest).from_(gateway)
6
7        print(f"Called URL: {mock.call_args[1]['url']}")
8        self.assertEqual(
9            mock.call_args[1]["url"],
10            f"{gateway.settings.server_url}/rates"  # Adapt to actual endpoint
11        )

2. Authentication Header Test

1def test_authentication_headers(self):
2    """Verify correct authentication headers are sent."""
3    with patch("karrio.mappers.[carrier_id].proxy.lib.request") as mock:
4        mock.return_value = "{}"
5        karrio.Rating.fetch(self.RateRequest).from_(gateway)
6
7        headers = mock.call_args[1]["headers"]
8        print(f"Headers sent: {headers}")
9        self.assertIn("Authorization", headers)
10        # For API key auth:
11        self.assertTrue(headers["Authorization"].startswith("Bearer "))
12        # For Basic auth:
13        # self.assertTrue(headers["Authorization"].startswith("Basic "))

3. Request Data Format Test

1def test_request_data_format(self):
2    """Verify request data is properly formatted."""
3    with patch("karrio.mappers.[carrier_id].proxy.lib.request") as mock:
4        mock.return_value = "{}"
5        karrio.Rating.fetch(self.RateRequest).from_(gateway)
6
7        sent_data = mock.call_args[1]["data"]
8        print(f"Sent data: {sent_data}")
9
10        # For JSON APIs:
11        self.assertIsInstance(sent_data, str)
12        parsed_data = lib.to_dict(sent_data)
13        self.assertIn("shipper", parsed_data)  # Verify structure
14
15        # For XML APIs:
16        # self.assertTrue(sent_data.startswith("<?xml"))

4. Concurrent Request Test (for tracking)

1def test_concurrent_tracking_requests(self):
2    """Test concurrent tracking requests work correctly."""
3    with patch("karrio.mappers.[carrier_id].proxy.lib.request") as mock:
4        mock.return_value = '{"status": "delivered"}'
5
6        request = models.TrackingRequest(tracking_numbers=["123", "456"])
7        response = karrio.Tracking.fetch(request).from_(gateway).parse()
8
9        print(f"Number of API calls: {mock.call_count}")
10        # Verify one call per tracking number
11        self.assertEqual(mock.call_count, len(request.tracking_numbers))

5. Development Stub Testing

During development, test with stubbed responses:

1def test_stubbed_rates_response(self):
2    """Test parsing works with stubbed response (development phase)."""
3    # Use stubbed response during development
4    stubbed_response = {
5        "rates": [{"serviceCode": "std", "totalCharge": 10.50}]
6    }
7
8    with patch("karrio.mappers.[carrier_id].proxy.lib.request") as mock:
9        mock.return_value = lib.to_json(stubbed_response)
10        response = karrio.Rating.fetch(self.RateRequest).from_(gateway).parse()
11
12        print(f"Stubbed response parsed: {lib.to_dict(response)}")
13        self.assertEqual(len(response[0]), 1)  # One rate returned
14        self.assertEqual(response[0][0].service, "std")

Proxy Testing Rules

  • Mock all HTTP calls: Never make real API calls in tests
  • Test exact URLs: Verify endpoints match carrier documentation
  • Verify authentication: Ensure correct headers are sent
  • Test data format: Verify JSON/XML structure is correct
  • Use stubbed data: Test with realistic carrier response formats
  • Debug prints: Always add print statements before assertions

The Proxy class forms the communication bridge between Karrio’s unified interface and the carrier’s specific API, handling all the technical details of HTTP communication while keeping business logic in the provider layer.