Skip to main content
Version: V3

Local Adapters

V3 route​

Prepare a configuration package and register it. Then create and review a plan before applying the reviewed plan or running a confirmed sync; Run a sync has the canonical procedure. The custom-adapter example package is examples/custom_adapter/package.yml, which is a shape reference and a source fixture rather than a package that runs register-to-apply as shipped.

Using local adapters​

Custom adapters cannot be registered at this release

Installing your adapter is not sufficient to make a package naming it registrable. Registration resolves the package's source.name against a closed set of capability declarations, and only the adapters bundled with infrahub-sync have one. A package naming any other adapter is refused at registration, and no configuration or version row is created.

The refusal you receive is the service's fixed envelope — HTTP 422, code configs-validation, family validation, no reason, and the message the configuration service refused the request. The underlying cause is an internal finding, missing-adapter at /configuration/source, which the refusal does not expose — and because the package is never stored, configs validate cannot report it either. No public diagnostic names the cause; that missing diagnostic is part of the gap. The attempt still leaves durable evidence: its idempotency receipt is reserved and then released, and an audit event is recorded with outcome unavailable.

An installed dotted import target or entry point does not change that outcome, because the refusal is about the missing capability declaration rather than about whether the class can be imported. The supported route today is to add the adapter to this repository together with its declaration — see Adding an adapter, which records the reproduction and the exact refusing surface.

The material below describes how adapter targets are resolved, and remains accurate for development work and for the bundled adapters.

Adapter identities a registered package may name​

  1. Built-ins: Adapters that ship with infrahub-sync (infrahub_sync.adapters.<name>). These are the only ones with a capability declaration, so in practice they are the only ones a registered package can use today.
  2. Dotted paths: Python module paths (myproj.adapters.foo:MyAdapter). Accepted by the envelope as a target, but not sufficient on its own — see the warning above.
  3. Python entry points: Installed packages that register entry points. Same limitation.

Filesystem paths, including .py targets, are refused in registered packages. The adapters_path configuration field is also refused.

The destination must name a bundled adapter that supports the destination role.

Development plugin-loader behavior​

The plugin loader can use local filesystem plugins while developing an adapter. INFRAHUB_SYNC_ADAPTER_PATHS configures that development behavior; it is not a registered-worker execution path, and it does not make a package registrable.

Creating a custom adapter​

A minimal custom adapter needs to extend diffsync.Adapter and implement the necessary methods:

from diffsync import Adapter

from infrahub_sync import DiffSyncMixin

class MyCustomAdapter(DiffSyncMixin, Adapter):
def __init__(self, target, adapter, config, *args, **kwargs):
super().__init__(*args, **kwargs)
self.target = target
self.settings = adapter.settings or {}
self.config = config

def model_loader(self, model_name, model):
# Your implementation to load data into the model
pass

For a more complete example, refer to examples/custom_adapter/package.yml and the repository's Custom Adapter Example. Its repository-local adapter target is development material, not an installed identity that a deployed worker can register and run as-is; and as the warning above records, installing it elsewhere would not make it registrable either.

Custom Adapter Examplehttps://github.com/opsmill/infrahub-sync/tree/main/examples/custom_adapter

Adding custom Jinja filters​

Custom adapters can provide their own Jinja filters for use in transform expressions. This is particularly useful for adapter-specific data transformations.

Implementing custom filters​

To add custom filters to your adapter, implement the _add_custom_filters class method in your DiffSync model class:

from typing import Any, ClassVar
from diffsync import DiffSyncModel
from infrahub_sync import DiffSyncModelMixin

class MyCustomModel(DiffSyncModelMixin, DiffSyncModel):
# Store any data needed by filters as class variables
_my_mapping: ClassVar[dict[str, str]] = {}

@classmethod
def set_my_mapping(cls, mapping: dict[str, str]) -> None:
"""Set mapping data for use in filters."""
cls._my_mapping = mapping

@classmethod
def _add_custom_filters(cls, native_env: Any, item: dict[str, Any]) -> None:
"""Add custom filters to the Jinja environment."""

def my_custom_filter(value: str) -> str:
"""Custom filter that transforms values using stored mapping."""
return cls._my_mapping.get(str(value), value)

def format_identifier(value: str) -> str:
"""Another custom filter for formatting identifiers."""
return f"ID-{value.upper()}"

# Register filters with the Jinja environment
native_env.filters["my_custom_filter"] = my_custom_filter
native_env.filters["format_identifier"] = format_identifier

Setting up filter data​

If your filters need data (like mappings, lookup values, etc.), initialize it in your adapter:

class MyCustomAdapter(DiffSyncMixin, Adapter):
def __init__(self, target, adapter, config, *args, **kwargs):
super().__init__(*args, **kwargs)
# ... other initialization

# Build data needed by filters
my_mapping = self._build_custom_mapping()

# Pass data to model class for filter use
MyCustomModel.set_my_mapping(my_mapping)

def _build_custom_mapping(self) -> dict[str, str]:
"""Build mapping data from your data source."""
# Implementation depends on your data source
return {"key1": "value1", "key2": "value2"}

Using custom filters in configuration​

Once implemented, use your custom filters in transform expressions:

schema_mapping:
- name: MyModel
mapping: "api/endpoint"
fields:
- name: identifier
mapping: "raw_id"
- name: formatted_name
mapping: "name"
transforms:
- field: identifier
expression: "{{ raw_id | my_custom_filter | format_identifier }}"
- field: status
expression: "{{ 'active' if enabled else 'inactive' }}"

Filter implementation guidelines​

  1. Keep filters focused: Each filter should do one specific transformation
  2. Handle edge cases: Always provide fallback values for missing data
  3. Use class variables: Store filter data as class variables for efficient access
  4. Document your filters: Add Python documentation strings explaining what each filter does
  5. Test thoroughly: Ensure filters work with various input types and edge cases

Real-world example: ACI device name filter​

Here's how the ACI adapter implements the aci_device_name filter:

class AciModel(DiffSyncModelMixin, DiffSyncModel):
_device_mapping: ClassVar[dict[str, str]] = {}

@classmethod
def set_device_mapping(cls, device_mapping: dict[str, str]) -> None:
cls._device_mapping = device_mapping

@classmethod
def _add_custom_filters(cls, native_env: Any, item: dict[str, Any]) -> None:
def aci_device_name(node_id: str) -> str:
"""Resolve ACI node IDs to device names."""
return cls._device_mapping.get(str(node_id), node_id)

native_env.filters["aci_device_name"] = aci_device_name

Used in configuration:

transforms:
- field: device
expression: "{{ l1PhysIf.attributes.dn.split('/')[2].replace('node-', '') | aci_device_name }}"

Best practices​

  1. Package Structure: Organize complex adapters as packages with __init__.py
  2. Testing: Include test data and documentation with your adapter
  3. Configuration: Use settings to make your adapter configurable
  4. Error Handling: Implement proper error handling and logging
  5. Type Annotations: Use type hints to make your code more maintainable
  6. Custom Filters: Implement adapter-specific Jinja filters for complex transformations