internal: Expose a Python API for Sync

[minitriga]

Have the ability to expose a python API so users can be able to start a sync within python without using the CLI.

[jjeff07]

My example code:

from infrahub_sync.utils import (
    get_instance,
    get_potenda_from_instance,
)
import sys
from pathlib import Path


def purge_adapter_modules(config_dir: str, adapters: list[str]) -> None:
    """
        Drop any sys.modules entries that could collide with the next sync's
        generated adapter (e.g. `infrahub`, `infrahub.adapter`,
    `infrahub.sync_models`).
        Also remove the sync directory from sys.path so the next sync's copy wins.
    """
    for adapter in adapters:
        for key in sys.modules:
            if key == adapter or key.startswith(f"{adapter}."):
                sys.modules.pop(key, None)

    dir_path = Path(__file__).resolve().parent
    sync_path = str(dir_path / config_dir)
    while sync_path in sys.path:
        sys.path.remove(sync_path)
            

_aruba_sync_instance = get_instance(
    config_file="config.yml",
    directory="aruba_sync",
)

_aruba_ptd = get_potenda_from_instance(
    sync_instance=_aruba_sync_instance,
    branch="main",
    show_progress=False,
)

_aruba_ptd.source_load()

_aruba_ptd.destination_load()

diff = _aruba_ptd.diff()

_aruba_ptd.sync(diff=diff)

print(diff.summary())

purge_adapter_modules(config_dir="aruba_sync", adapters=["infrahub", "custom"])

More complete example. Needs some work still:

from __future__ import annotations

import sys
from pathlib import Path
from typing import TYPE_CHECKING, Any

from infrahub_sync.utils import get_instance, get_potenda_from_instance, find_missing_schema_model, render_adapter
from infrahub_sdk import InfrahubClientSync, Config
from pydantic import BaseModel

if TYPE_CHECKING:
    from diffsync.diff import Diff
    from infrahub_sync import SyncInstance, Potenda


class InfrahbubSync(BaseModel):
    name: str | None = None
    config_file: str | None = "config.yml"
    directory: str | None = None
    branch: str | None = None
    show_progress: bool | None = None
    verbosity: int | None = None
    _sync_instance: SyncInstance
    _ptd: Potenda
    _diff: Diff | None = None
    _infra_config: Config

    def model_post_init(self, context: Any, /) -> None:
        sync_instance = get_instance(
            name=self.name,
            config_file=self.config_file,
            directory=self.directory,
        )
        if not sync_instance:
            raise ValueError("No sync instance found with the provided parameters.")

        self._sync_instance = sync_instance
        self._ptd = get_potenda_from_instance(
            sync_instance=self._sync_instance,
            branch=self.branch,
            show_progress=self.show_progress,
            verbosity=self.verbosity,
        )
        self._check_adapters()

    @property
    def diff(self) -> Diff | None:
        return self._diff

    def purge_adapter_modules(self) -> None:
        """
            Drop any sys.modules entries that could collide with the next sync's
            generated adapter (e.g. `infrahub`, `infrahub.adapter`,
        `infrahub.sync_models`).
            Also remove the sync directory from sys.path so the next sync's copy wins.
        """

        for key in sys.modules:
            for adapter in (self._sync_instance.source.name, self._sync_instance.destination.name):
                if key == adapter or key.startswith(f"{adapter}."):
                    sys.modules.pop(key, None)

        sync_path = str(Path(self.directory).resolve())
        while sync_path in sys.path:
            sys.path.remove(sync_path)

    def create_diff(self) -> Diff:
        self._check_adapters()
        self._ptd.source_load()
        self._ptd.destination_load()
        _ = self._ptd.diff()
        self._diff = _
        return _

    def sync(self) -> Diff:
        if not self._diff:
            self.create_diff()
        diff = self._ptd.sync(diff=self._diff)
        return diff

    def generate(self):
        # Initialize InfrahubClientSync if address and config are available
        infra_settings = None
        for adapter in (self._sync_instance.source, self._sync_instance.destination):
            if adapter.name == "infrahub":
                infra_settings = adapter.settings
                break
        if not infra_settings:
            raise ValueError("Infrahub configuration is missing or incomplete.")
        # TODO: Need to handle merging config settings with ENV vars and defaults
        client = InfrahubClientSync(config=infra_settings)
        schema = client.schema.all()

        if _ := find_missing_schema_model(sync_instance=self._sync_instance, schema=schema):
            raise ValueError(f"One or more models are not present in the Schema - {_}")

        for template, output_path in render_adapter(
            sync_instance=self._sync_instance, schema=schema
        ):
            print(f"Rendered Infrahub Sync template files: {template} -> {output_path}")

    def _check_adapters(self):
        for sync_name in [self._sync_instance.source.name, self._sync_instance.destination.name]:
            for file in ["sync_models.py", "sync_adapter.py"]:
                if not (Path(self.directory) /  file).exists():
                    print(
                        f"Missing {sync_name} adapter files; generating new ones."
                    )
                    self.generate()

sync = InfrahbubSync(name="from-ipfabric", show_progress=False, directory="src/ipf_sync")
diff = sync.create_diff()
print(diff.summary())

[BeArchiTek]

Thanks for sharing this @jjeff07