Building a Vendor-Neutral Telematics Adapter Layer

This guide implements the adapter layer concretely: an abstract base contract, a registry keyed by a vendor enum, per-vendor concrete classes for Samsara, Geotab, and Motive, and dependency-injected config, arranged so the dispatcher selects an adapter by enum and a fourth vendor can be added without touching the core. The payoff is architectural insurance — when a new customer arrives on a platform you have not integrated, or a vendor changes its auth, you write one class instead of editing the delivery path that carries out-of-service notifications. Get the boundaries wrong and vendor-specific code leaks upward, and every OOS event under 49 CFR § 396.9 risks a delivery path that only one team member understands.

This is the implementation beneath Samsara, Geotab and Motive Platform Adapters, inside the Telematics, ELD & Maintenance Platform Integration section that defines the DvirComplianceEvent these adapters deliver.

Prerequisites

Python 3.10+ — the code uses Protocol, abc, match/case, and X | Y unions.
Pydantic 2.x — per-vendor AdapterConfig with a SecretStr credential.
pytest — the registry and protocol-conformance tests below.
The Vendor enum and DvirComplianceEvent — the routing key and canonical event from the parent guide and section.

Step-by-Step Implementation

The layer is a base class, a registry, three concretes, and a dispatcher. Build it so the only vendor-aware code lives inside the concrete classes.

Step 1 — Define the abstract base with an ABC

Use an abstract base class so every adapter is forced to implement the full contract; an incomplete adapter fails to instantiate rather than exploding at delivery time:

from abc import ABC, abstractmethod
from enum import Enum

from pydantic import BaseModel, SecretStr


class Vendor(str, Enum):
    SAMSARA = "samsara"
    GEOTAB = "geotab"
    MOTIVE = "motive"


class AdapterConfig(BaseModel):
    vendor: Vendor
    api_base_url: str
    credential: SecretStr
    rate_limit_per_min: int
    ack_mode: str = "sync"


class DeliveryAck(BaseModel):
    vendor: Vendor
    accepted: bool
    vendor_ref: str | None = None
    deferred: bool = False


class TelematicsAdapter(ABC):
    vendor: Vendor

    def __init__(self, config: AdapterConfig) -> None:
        self.config = config  # dependency-injected; no hard-coded credentials

    @abstractmethod
    async def authenticate(self) -> None: ...

    @abstractmethod
    async def resolve_vehicle_id(self, vehicle_vin: str) -> str: ...

    @abstractmethod
    async def push_event(self, event: dict) -> DeliveryAck: ...

Step 2 — Implement per-vendor concrete adapters

Each concrete class translates the internal event into one vendor’s auth, id scheme, and payload. The stubs below show the shape; the vendor-specific translation is the only place a vendor name appears:

class SamsaraAdapter(TelematicsAdapter):
    vendor = Vendor.SAMSARA

    async def authenticate(self) -> None:
        # Bearer token — nothing to refresh; validate presence.
        assert self.config.credential.get_secret_value()

    async def resolve_vehicle_id(self, vehicle_vin: str) -> str:
        return await self._lookup_asset_id(vehicle_vin)  # numeric asset id

    async def push_event(self, event: dict) -> DeliveryAck:
        asset_id = await self.resolve_vehicle_id(event["vehicle_vin"])
        # ... POST flat JSON to the Samsara webhook endpoint ...
        return DeliveryAck(vendor=self.vendor, accepted=True, vendor_ref=asset_id)

    async def _lookup_asset_id(self, vin: str) -> str:
        return f"samsara-{vin[-6:]}"  # stub


class GeotabAdapter(TelematicsAdapter):
    vendor = Vendor.GEOTAB

    async def authenticate(self) -> None:
        # Session-based MyGeotab auth — acquire a session id.
        ...

    async def resolve_vehicle_id(self, vehicle_vin: str) -> str:
        return f"b{vehicle_vin[-4:]}"  # GUID-like Id string (stub)

    async def push_event(self, event: dict) -> DeliveryAck:
        # Poll-based vendor: enqueue for the next GetFeed cycle.
        return DeliveryAck(vendor=self.vendor, accepted=True, deferred=True)


class MotiveAdapter(TelematicsAdapter):
    vendor = Vendor.MOTIVE

    async def authenticate(self) -> None:
        # OAuth 2.0 — acquire or refresh an access token.
        ...

    async def resolve_vehicle_id(self, vehicle_vin: str) -> str:
        return f"motive-{vehicle_vin[-6:]}"  # numeric vehicle id (stub)

    async def push_event(self, event: dict) -> DeliveryAck:
        return DeliveryAck(vendor=self.vendor, accepted=True, vendor_ref="ok")

Step 3 — Build a registry keyed by the vendor enum

The registry maps each Vendor to its adapter class. Resolving an unregistered vendor raises, so an unknown vendor is a loud failure, not a dropped out-of-service event:

class UnknownVendorError(KeyError):
    """Raised when no adapter is registered for a vendor."""


_REGISTRY: dict[Vendor, type[TelematicsAdapter]] = {}


def register(cls: type[TelematicsAdapter]) -> type[TelematicsAdapter]:
    """Class decorator that registers an adapter under its vendor."""
    _REGISTRY[cls.vendor] = cls
    return cls


for _cls in (SamsaraAdapter, GeotabAdapter, MotiveAdapter):
    register(_cls)


def resolve_adapter(config: AdapterConfig) -> TelematicsAdapter:
    """Instantiate the adapter for config.vendor, or raise UnknownVendorError."""
    try:
        cls = _REGISTRY[config.vendor]
    except KeyError as exc:
        raise UnknownVendorError(f"no adapter registered for {config.vendor}") from exc
    return cls(config)

Step 4 — Dispatch through the registry

The dispatcher takes a fleet’s config and an event, resolves the adapter, and delegates. It contains no vendor names, so adding a vendor never changes it:

async def dispatch(config: AdapterConfig, event: dict) -> DeliveryAck:
    adapter = resolve_adapter(config)
    await adapter.authenticate()
    return await adapter.push_event(event)

Step 5 — Add a fourth vendor without touching the core

A new vendor is one enum value, one concrete class, and one registration — the dispatcher, registry, and base class are untouched:

# In the Vendor enum, add:  VERIZON_CONNECT = "verizon_connect"

@register
class VerizonConnectAdapter(TelematicsAdapter):
    vendor = Vendor.VERIZON_CONNECT

    async def authenticate(self) -> None: ...
    async def resolve_vehicle_id(self, vehicle_vin: str) -> str:
        return f"vzc-{vehicle_vin[-6:]}"
    async def push_event(self, event: dict) -> DeliveryAck:
        return DeliveryAck(vendor=self.vendor, accepted=True)

Verification and Testing

Assert three properties: the registry resolves the correct adapter, an unknown vendor raises, and every concrete adapter satisfies the protocol.

import pytest


def _cfg(vendor: Vendor) -> AdapterConfig:
    return AdapterConfig(vendor=vendor, api_base_url="https://x", credential="s",
                         rate_limit_per_min=60)


@pytest.mark.parametrize("vendor,expected", [
    (Vendor.SAMSARA, SamsaraAdapter),
    (Vendor.GEOTAB, GeotabAdapter),
    (Vendor.MOTIVE, MotiveAdapter),
])
def test_registry_resolves_correct_adapter(vendor, expected):
    adapter = resolve_adapter(_cfg(vendor))
    assert isinstance(adapter, expected)
    assert adapter.vendor is vendor


def test_unknown_vendor_raises():
    # A vendor value with no registered adapter must raise, never no-op.
    _REGISTRY.pop(Vendor.MOTIVE, None)
    with pytest.raises(UnknownVendorError):
        resolve_adapter(_cfg(Vendor.MOTIVE))
    register(MotiveAdapter)  # restore for other tests


@pytest.mark.parametrize("cls", [SamsaraAdapter, GeotabAdapter, MotiveAdapter])
def test_each_adapter_satisfies_protocol(cls):
    adapter = cls(_cfg(cls.vendor))
    for method in ("authenticate", "resolve_vehicle_id", "push_event"):
        assert callable(getattr(adapter, method))


@pytest.mark.asyncio
async def test_geotab_delivery_is_deferred():
    ack = await dispatch(_cfg(Vendor.GEOTAB), {"vehicle_vin": "1FUJA6", "event_id": "e1"})
    assert ack.deferred is True  # poll-based vendor never fakes a sync success

Common Failure Modes and Gotchas

Differing rate limits. Each vendor enforces its own request budget, and a shared limiter would let Samsara’s traffic starve Geotab’s or vice versa. Give every adapter its own limiter sized to that vendor’s budget, and inject it through AdapterConfig so the limit is configuration, not a constant buried in the class. When a limit is hit, queue with priority rather than dropping — an out-of-service event must never be discarded to satisfy a throttle.
Id-mapping drift. A cached VIN-to-native-id mapping goes stale when a vendor re-keys a vehicle after re-provisioning, and push_event then fails with an unknown-vehicle error. Do not trust the cache blindly: on a delivery failure that looks like an unknown vehicle, call resolve_vehicle_id again to refresh the mapping, and only dead-letter the event if the fresh lookup also fails. Treat the native id as a cache, never as a permanent key.
Partial vendor outages. When one vendor’s API is down, the registry must keep delivering to the others; a single authenticate failure should not take down the dispatcher. Wrap each adapter call so a failing vendor routes its events to a dead-letter queue for replay while healthy vendors continue. Never let one vendor’s outage raise out of dispatch and abort a batch destined for several platforms.

Frequently Asked Questions

Should the base be a Protocol or an ABC?

Use an ABC when you want every adapter forced to implement the full contract at instantiation — an incomplete concrete class raises TypeError when constructed, which fails fast. A Protocol is better when adapters come from code you do not control and you only need structural conformance checked at the boundary. This layer uses an ABC because you own every adapter and want the compiler-like guarantee that none ships half-implemented.

How does adding a vendor avoid touching the dispatcher?

Because the dispatcher resolves adapters through the registry rather than branching on vendor name. A new vendor is one enum value, one concrete class, and one @register call; the registry picks it up and the dispatcher delegates to it unchanged. The absence of any if vendor == ... logic above the adapter is what keeps the addition local.

What should resolving an unregistered vendor do?

Raise. A silent no-op or a None return would drop the event, and if that event is an out-of-service notification the vehicle keeps operating without the platform ever being told. resolve_adapter raises UnknownVendorError so an unconfigured vendor surfaces immediately as a loud failure that can be alerted on, not a quiet gap in delivery.

Samsara, Geotab and Motive Platform Adapters — the vendor comparison and protocol design this implements.
Webhook Event Delivery for DVIR Pipelines — the delivery layer the dispatcher feeds.
Idempotent Webhook Retry and Signature Verification — the retry and rate-limit guarantees around delivery.
Designing a Dead-Letter Queue for Failed DVIR Payloads — where a failing adapter’s events wait for replay.

This guide is part of Samsara, Geotab and Motive Platform Adapters. Back to the Telematics, ELD & Maintenance Platform Integration section.