Skip to content

Architecture Deep Dive

Advanced topic

This page is for SDK contributors and developers who want to understand the internal machinery. You do not need this to use proxies, plugins, or capabilities in your function blocks.

This page covers the metaclass, method delegation, fallback generation, and registry internals of the connection proxy system.

Three-Layer Architecture

flowchart TB
    subgraph "User Code"
        FB["Function Block"]
    end

    subgraph "Proxy Layer"
        Proxy["ConnectionProxy subclass"]
        Meta["ProxyMeta metaclass"]
    end

    subgraph "Plugin Layer"
        Plugin["ConnectionPlugin"]
        Cap["Capability Interfaces"]
    end

    subgraph "Connection Layer"
        BC["BaseConnection"]
        Raw["Raw Library Client<br/>(Scrapli / Netmiko / httpx)"]
    end

    FB --> Proxy
    Meta -.->|"generates fallbacks"| Proxy
    Proxy -->|"delegates via __getattribute__"| Plugin
    Plugin -->|"implements"| Cap
    Plugin -->|"owns"| BC
    BC -->|"wraps"| Raw
Layer Responsibility
Proxy Typed API surface, method delegation, fallback generation
Plugin Platform-specific logic, capability implementations, connection factory
Connection Library wrapper, connect/disconnect/is_alive lifecycle

ProxyMeta Metaclass

ProxyMeta extends ABCMeta and runs at class creation time for every ConnectionProxy subclass. It performs four operations:

1. Collect Capability Interfaces

The metaclass walks the MRO and identifies every base that is a subclass of CapabilityInterface (excluding CapabilityInterface itself and other proxy classes). These are stored in cls._capability_interfaces.

2. Collect Abstract Methods

For each capability interface, the metaclass collects all names in __abstractmethods__ and maps them to their source interface in cls._capability_methods.

3. Generate Fallback Methods

For each abstract method not already defined in the class namespace, the metaclass generates a concrete fallback. The fallback:

  • Reads self._device and self._plugin at call time to build a CapabilityContext.
  • Raises NotImplementedForThisPlatform with the context attached.
  • Preserves the original method's __name__, __qualname__, __doc__, and __annotations__.
  • Is tagged with __generated_fallback__ = True for introspection.

4. Clear Abstract Methods

The metaclass removes all capability method names from __abstractmethods__, making the proxy class instantiable. Static analyzers may still warn about abstract classes -- this is a known IDE limitation.

Method Delegation via __getattribute__

ConnectionProxy.__getattribute__ intercepts every attribute access:

flowchart LR
    Access["proxy.method()"] --> Private{"Starts with<br/>'_'?"}
    Private -- Yes --> Self1["Return from self"]
    Private -- No --> Cap{"In<br/>_capability_methods?"}
    Cap -- Yes --> PluginCheck{"Plugin has<br/>callable attr?"}
    PluginCheck -- Yes --> Delegate["Return plugin.method"]
    PluginCheck -- No --> Self2["Return from self<br/>(fallback)"]
    Cap -- No --> Self2

The resolution priority is:

  1. Private attributes (_device, _plugin, _closed, ...) -- always resolved from the proxy instance via object.__getattribute__.
  2. Capability methods -- if the method name exists in _capability_methods, the proxy checks the plugin for a callable attribute with that name. If found, the plugin's method is returned directly.
  3. Self -- everything else (including metaclass-generated fallbacks, explicit proxy overrides, and non-capability methods) resolves from the proxy instance.

This means a user can override a fallback by defining the method on the proxy class itself:

class SafeProxy(ConnectionProxy, DeviceInfoCapability):
    def get_version(self) -> dict[str, str | None]:
        return {"vendor": None, "model": None, "serial": None, "software_release": None}

The override only runs when the plugin does not implement the method.

Fallback Method Internals

Each generated fallback captures three pieces of context at creation time (closure variables):

  • cls -- the proxy class being created
  • method_name -- the string name of the method
  • interface -- the capability interface that declares it

At call time, the fallback reads self._device and self._plugin to populate the CapabilityContext dataclass, then raises NotImplementedForThisPlatform:

raise NotImplementedForThisPlatform(
    f"Method '{method_name}()' is not implemented for this platform.",
    context=CapabilityContext(
        proxy_class=cls.__name__,
        method_name=method_name,
        interface_name=interface.__name__,
        platform=...,
        plugin_class=...,
        device_id=...,
    ),
)

The exception includes enough information for callers to diagnose the issue without inspecting the stack trace.

Registry Internals

The plugin registry is a set of module-level dictionaries:

Structure Type Purpose
_plugin_registry dict[platform, dict[type, dict[library, class]]] Main three-level lookup
_platform_defaults dict[platform, class] One default per platform
_connection_type_defaults dict[(platform, type), class] One default per platform+type

resolve_plugin() Algorithm (Detail)

  1. Extract platform from device.platform.short_name or device.platform (string). Raise ConnectionValidationError if missing.
  2. Filter _plugin_registry by platform, then optionally by connection_type and connection_library. Result: a flat list of candidate classes.
  3. Capability filtering -- if required_capabilities is non-empty:
    • Compute full matches (plugins implementing all capabilities).
    • If full matches exist, replace candidates with them.
    • Otherwise, score each candidate by number of capabilities implemented. Select the highest scorer. Ties raise AmbiguousPluginError.
  4. Single candidate -- instantiate with plugin_class(device=device) and return.
  5. Multiple candidates with fallback_to_default=True:
    • Check _connection_type_defaults[(platform, connection_type)] first.
    • Then check _platform_defaults[platform].
    • The default must be in the candidate list to be selected.
  6. Multiple candidates without a usable default -- raise AmbiguousPluginError with full diagnostic information.

SDK vs User Plugin Detection

The registry determines plugin origin by comparing the plugin's module root to the SDK's module root (neops_worker_sdk). This is derived at import time from ConnectionPlugin.__module__. User plugins defined in any other package always take precedence for default registration.

Next Steps