Core Classes

class Api:
    """The API object is the point of entry to pynetbox.

    After instantiating the Api() with the appropriate named arguments
    you can specify which app and endpoint you wish to interact with.

    Valid attributes currently are:

    * circuits
    * core (NetBox 3.5+)
    * dcim
    * extras
    * ipam
    * tenancy
    * users
    * virtualization
    * vpn (NetBox 3.7+)
    * wireless

    Calling any of these attributes will return an `App` object which exposes endpoints as attributes.

    ## Additional Attributes

    * **http_session(requests.Session)**: Override the default session with your own. This is used to control
      a number of HTTP behaviors such as SSL verification, custom headers,
      retires, and timeouts.
      See [custom sessions](advanced.md#custom-sessions) for more info.

    ## Parameters

    * **url** (str): The base URL to the instance of NetBox you wish to connect to.
    * **token** (str): Your NetBox token.
    * **threading** (bool, optional): Set to True to use threading in `.all()` and `.filter()` requests.
    * **thread_pool_executor** (callable, optional): A `concurrent.futures.ThreadPoolExecutor`
      class (or any callable matching its `(max_workers=...)` signature and context-manager
      protocol) used to build the pool for threaded requests. Defaults to
      `concurrent.futures.ThreadPoolExecutor`. Inject a custom executor to propagate
      thread-local state such as OpenTelemetry trace context into worker threads.
    * **max_workers** (int, optional): Maximum number of worker threads used for threaded
      `.all()` and `.filter()` requests. Defaults to 4. Only used when `threading=True`.

    ## Raises

    * **AttributeError**: If app doesn't exist.

    ## Examples

    ```python
    import pynetbox
    nb = pynetbox.api(
        'http://localhost:8000',
        token='d6f4e314a5b5fefd164995169f28ae32d987704f'
    )
    list(nb.dcim.devices.all())
    # [test1-leaf1, test1-leaf2, test1-leaf3]
    ```
    """

    def __init__(
        self,
        url,
        token=None,
        threading=False,
        strict_filters=False,
        extensions=None,
        pagination="offset",
        thread_pool_executor=None,
        max_workers=4,
    ):
        """Initialize the API client.

        Args:
            url (str): The base URL to the instance of NetBox you wish to connect to.
            token (str, optional): Your NetBox API token. If not provided, authentication will be required for each
                request.
            threading (bool, optional): Set to True to use threading in `.all()` and `.filter()` requests, defaults to
                False.
            strict_filters (bool, optional): Set to True to check GET call filters against OpenAPI specifications
                (intentionally not done in NetBox API), defaults to False.
            extensions (list, optional): A list of `Extension` classes or instances that register custom `Record`
                subclasses and content-type mappings for NetBox plugins. See `pynetbox.core.extension`.
            pagination (str, optional): Pagination strategy for `.all()` and `.filter()`, either `"offset"` (default) or
                `"cursor"`. Cursor pagination (NetBox 4.6+) offers better performance on very large result sets but
                omits the total count and cannot be combined with threading or `ordering`. On NetBox versions older than
                4.6 it transparently falls back to offset pagination.
            thread_pool_executor (callable, optional): A `concurrent.futures.ThreadPoolExecutor` class, or any callable
                matching its `(max_workers=...)` signature and context-manager protocol, used to build the pool for
                threaded requests. Defaults to `concurrent.futures.ThreadPoolExecutor`.
            max_workers (int, optional): Maximum number of worker threads used for threaded requests, defaults to 4.
        """
        if pagination not in ("offset", "cursor"):
            raise ValueError(
                "pagination must be 'offset' or 'cursor', got {!r}".format(pagination)
            )
        if max_workers <= 0:
            raise ValueError("max_workers must be a positive integer")

        base_url = "{}/api".format(url if url[-1] != "/" else url[:-1])
        self.token = token
        self.base_url = base_url
        self.http_session = requests.Session()
        self.threading = threading
        # Stored as ``None`` (the sentinel) when the caller did not supply a
        # custom executor, so ``None`` distinguishes "use the default" from an
        # explicit choice. ``Request`` resolves ``None`` to
        # ``concurrent.futures.ThreadPoolExecutor`` at request time, so the
        # value here is not necessarily the executor actually used.
        self.thread_pool_executor = thread_pool_executor
        self.max_workers = max_workers
        self.strict_filters = strict_filters
        self.pagination = pagination
        self._cursor_supported = None

        self._register_extensions(extensions or [])

        # Initialize NetBox apps
        self.circuits = App(self, "circuits")
        self.core = App(self, "core")
        self.dcim = App(self, "dcim")
        self.extras = App(self, "extras")
        self.ipam = App(self, "ipam")
        self.tenancy = App(self, "tenancy")
        self.users = App(self, "users")
        self.virtualization = App(self, "virtualization")
        self.vpn = App(self, "vpn")
        self.wireless = App(self, "wireless")
        self.plugins = PluginsApp(self)

    def _register_extensions(self, extensions):
        """Build per-instance extension and content-type registries.

        Stores extensions keyed by ``plugin_name`` so ``PluginsApp`` and
        ``App._setmodel`` can look up the right ``models`` namespace, and
        composes the built-in ``CONTENT_TYPE_MAPPER`` with each extension's
        ``content_types`` overrides.

        Two extensions cannot share a ``plugin_name`` or register the same
        content-type key — both raise ``ValueError``, since a silent
        last-wins would mask a configuration mistake. Overriding a key from
        the built-in ``CONTENT_TYPE_MAPPER`` is allowed and intentional.

        ``plugin_name`` is normalized by converting dashes to underscores
        before lookup so the same plugin cannot register itself twice
        under both spellings (e.g. ``"custom-objects"`` and
        ``"custom_objects"``).
        """
        self._extensions = {}
        for ext in extensions:
            plugin_name = getattr(ext, "plugin_name", None)
            if not plugin_name:
                raise ValueError(
                    "Extension {!r} is missing a 'plugin_name' attribute".format(ext)
                )
            plugin_name = plugin_name.replace("-", "_")
            if plugin_name in self._extensions:
                raise ValueError(
                    "Duplicate extension for plugin_name {!r}: {!r} and {!r}".format(
                        plugin_name, self._extensions[plugin_name], ext
                    )
                )
            self._extensions[plugin_name] = ext

        content_types = dict(CONTENT_TYPE_MAPPER)
        seen_extension_keys = {}
        for ext in self._extensions.values():
            ext_content_types = getattr(ext, "content_types", None) or {}
            for key, value in ext_content_types.items():
                if key in seen_extension_keys:
                    raise ValueError(
                        "Duplicate content_type {!r} registered by extensions "
                        "{!r} and {!r}".format(key, seen_extension_keys[key], ext)
                    )
                seen_extension_keys[key] = ext
                content_types[key] = value
        self._content_type_mapper = content_types
        self._type_content_mapper = {
            v: k for k, v in content_types.items() if v is not None
        }

    @property
    def version(self):
        """Gets the API version of NetBox.

        Can be used to check the NetBox API version if there are
        version-dependent features or syntaxes in the API.

        ## Returns
        Version number as a string.

        ## Example

        ```python
        import pynetbox
        nb = pynetbox.api(
            'http://localhost:8000',
            token='d6f4e314a5b5fefd164995169f28ae32d987704f'
        )
        nb.version
        # '3.1'
        ```
        """
        version = Request(
            base=self.base_url,
            token=self.token,
            http_session=self.http_session,
        ).get_version()
        return version

    def _effective_pagination(self):
        """Resolve the pagination strategy to use for list requests.

        Returns ``"cursor"`` only when cursor pagination was requested *and*
        the connected NetBox supports it (4.6+). Otherwise returns
        ``"offset"``. The server version is probed once and cached, so only
        the first `.all()`/`.filter()` on a cursor-mode `Api` pays the cost;
        offset-mode instances never make the extra request.
        """
        if self.pagination != "cursor":
            return "offset"
        if self._cursor_supported is None:
            try:
                self._cursor_supported = version.parse(self.version) >= version.parse(
                    "4.6"
                )
            except (RequestError, InvalidVersion, requests.exceptions.RequestException):
                # RequestError covers a non-ok HTTP response from the version
                # probe; requests.exceptions.RequestException covers transport
                # failures (ConnectionError, Timeout, ...) raised before a
                # response exists. In every case fall back to offset, as the
                # docstring promises, and let the real list request surface any
                # underlying connectivity error.
                self._cursor_supported = False
            if self._cursor_supported and self.threading:
                # Cursor pagination follows next links sequentially and cannot
                # be parallelised; the cursor path ignores self.threading.
                # Warn so the no-op threading configuration is not a silent
                # performance surprise.
                # stacklevel=5 attributes the warning to the caller's list
                # request rather than to pynetbox internals. The version probe
                # is resolved lazily on the first page fetch, so the fixed
                # frame chain at this point is:
                #   _effective_pagination -> Request._resolve_pagination
                #   -> Request.get -> RecordSet.__next__/__len__ -> caller.
                warnings.warn(
                    "threading=True has no effect with cursor pagination; "
                    "cursor pages are fetched sequentially.",
                    stacklevel=5,
                )
        return "cursor" if self._cursor_supported else "offset"

    def openapi(self):
        """Returns the OpenAPI spec.

        Quick helper function to pull down the entire OpenAPI spec.
        It is stored in memory to avoid repeated calls on NetBox API.

        ## Returns
        dict: The OpenAPI specification as a dictionary.

        ## Example

        ```python
        import pynetbox
        nb = pynetbox.api(
            'http://localhost:8000',
            token='d6f4e314a5b5fefd164995169f28ae32d987704f'
        )
        nb.openapi()
        # {...}
        ```
        """
        if not (openapi := getattr(self, "_openapi", None)):
            openapi = self._openapi = Request(
                base=self.base_url,
                http_session=self.http_session,
            ).get_openapi()

        return openapi

    def status(self):
        """Gets the status information from NetBox.

        ## Returns
        Dictionary containing NetBox status information.

        ## Raises
        `RequestError`: If the request is not successful.

        ## Example

        ```python
        from pprint import pprint
        pprint(nb.status())
        {
            'django-version': '3.1.3',
            'installed-apps': {
                'cacheops': '5.0.1',
                'debug_toolbar': '3.1.1',
                'django_filters': '2.4.0',
                'django_prometheus': '2.1.0',
                'django_rq': '2.4.0',
                'django_tables2': '2.3.3',
                'drf_yasg': '1.20.0',
                'mptt': '0.11.0',
                'rest_framework': '3.12.2',
                'taggit': '1.3.0',
                'timezone_field': '4.0'
            },
            'netbox-version': '2.10.2',
            'plugins': {},
            'python-version': '3.7.3',
            'rq-workers-running': 1
        }
        ```
        """
        status = Request(
            base=self.base_url,
            token=self.token,
            http_session=self.http_session,
        ).get_status()
        return status

    def create_token(self, username, password):
        """Creates an API token using a valid NetBox username and password.
        Saves the created token automatically in the API object.

        ## Parameters
        * **username** (str): NetBox username
        * **password** (str): NetBox password

        ## Returns
        `Record`: The token as a Record object.

        ## Raises
        `RequestError`: If the request is not successful.

        ## Notes

        NetBox 4.5 introduced v2 tokens. For v2 tokens, `nb.token` is set to
        `nbt_<key>.<token>` (the full auth value required in the Authorization
        header), which differs from `token.key`. For v1 tokens (pre-4.5),
        `nb.token` is the plaintext token value.

        ## Example

        ```python
        import pynetbox
        nb = pynetbox.api("https://netbox-server")
        token = nb.create_token("admin", "netboxpassword")

        # NetBox 4.5+ v2 token: nb.token differs from token.key
        nb.token
        # 'nbt_shortkey1234567.plaintexttoken7890abcdef1234567890abcdef'
        token.key
        # 'shortkey1234567'

        # Pre-4.5 / v1 token: nb.token matches token.key (or token.token)
        nb.token
        # '96d02e13e3f1fdcd8b4c089094c0191dcb045bef'
        ```
        """
        resp = Request(
            base="{}/users/tokens/provision/".format(self.base_url),
            http_session=self.http_session,
        ).post(data={"username": username, "password": password})
        # v2 tokens (NetBox 4.5+): construct auth value as nbt_<key>.<token>
        if resp.get("version") == 2:
            self.token = "{}{}.{}".format(TOKEN_PREFIX, resp["key"], resp["token"])
        else:
            self.token = resp.get("token") or resp["key"]
        return Record(resp, self, None)

    @contextlib.contextmanager
    def activate_branch(self, branch):
        """Context manager to activate the branch by setting the schema ID in the headers.

        **Note**: The NetBox branching plugin must be installed and enabled in your NetBox instance for this
        functionality to work.

        ## Parameters
        * **branch** (Record): The NetBox branch to activate

        ## Raises
        `ValueError`: If the branch is not a valid NetBox branch.

        ## Example

        ```python
        import pynetbox
        nb = pynetbox.api("https://netbox-server")
        branch = nb.plugins.branching.branches.create(name="testbranch")
        with nb.activate_branch(branch):
            sites = nb.dcim.sites.all()
            # All operations within this block will use the branch's schema
        ```
        """
        if not isinstance(branch, Record) or "schema_id" not in dict(branch):
            raise ValueError(
                f"The specified branch is not a valid NetBox branch: {branch}."
            )

        self.http_session.headers["X-NetBox-Branch"] = branch.schema_id

        try:
            yield
        finally:
            self.http_session.headers.pop("X-NetBox-Branch", None)

class App:
    """Represents apps in NetBox.

    Calls to attributes are returned as Endpoint objects.

    ## Returns
    Endpoint matching requested attribute.

    ## Raises
    RequestError if requested endpoint doesn't exist.
    """

    def __init__(self, api, name):
        self.api = api
        self.name = name
        self._setmodel()

    models = {
        "circuits": circuits,
        "core": core,
        "dcim": dcim,
        "extras": extras,
        "ipam": ipam,
        "users": users,
        "virtualization": virtualization,
        "wireless": wireless,
    }

    _PLUGINS_PREFIX = "plugins/"

    def _setmodel(self):
        if self.name.startswith(self._PLUGINS_PREFIX):
            # Plugin apps may carry a custom models namespace registered
            # via an Extension. The plugin attribute uses underscores while
            # the URL slug uses dashes (PluginsApp.__getattr__), so convert
            # back when looking up the registry.
            plugin_slug = self.name[len(self._PLUGINS_PREFIX) :]
            plugin_name = plugin_slug.replace("-", "_")
            extensions = getattr(self.api, "_extensions", {})
            ext = extensions.get(plugin_name)
            self.model = getattr(ext, "models", None) if ext is not None else None
        else:
            self.model = App.models.get(self.name)

    def __getstate__(self):
        return {"api": self.api, "name": self.name}

    def __setstate__(self, d):
        self.__dict__.update(d)
        self._setmodel()

    def __getattr__(self, name):
        return Endpoint(self.api, self, name, model=self.model)

    def endpoint(self, name):
        """Return an Endpoint using ``name`` as the literal URL slug.

        Attribute access (``app.ip_addresses``) converts underscores to
        dashes, which is correct for the vast majority of NetBox endpoints.
        This method skips that conversion so endpoints whose slug genuinely
        contains underscores can be reached.

        ## Parameters

        * **name** (str): The endpoint slug, used verbatim (no ``_`` → ``-``).

        ## Returns
        Endpoint matching the given slug.

        ## Examples

        ```python
        nb.plugins.custom_objects.endpoint("my_custom_object").all()
        ```
        """
        return Endpoint(self.api, self, name, model=self.model, literal_name=True)

    def config(self):
        """Returns config response from app.

        ## Returns
        Raw response from NetBox's config endpoint.

        ## Raises
        RequestError if called for an invalid endpoint.

        ## Examples

        ```python
        pprint.pprint(nb.users.config())
        {
            'tables': {
                'DeviceTable': {
                    'columns': [
                        'name',
                        'status',
                        'tenant',
                        'role',
                        'site',
                        'primary_ip',
                        'tags'
                    ]
                }
            }
        }
        ```
        """
        config = Request(
            base="{}/{}/config/".format(
                self.api.base_url,
                self.name,
            ),
            token=self.api.token,
            http_session=self.api.http_session,
        ).get()
        return config

class PluginsApp:
    """Basically valid plugins api could be handled by same App class,
    but you need to add plugins to request url path.

    ## Returns
    App with added plugins into path.
    """

    def __init__(self, api):
        self.api = api

    def __getstate__(self):
        return self.__dict__

    def __setstate__(self, d):
        self.__dict__.update(d)

    def __getattr__(self, name):
        return App(self.api, "plugins/{}".format(name.replace("_", "-")))

    def installed_plugins(self):
        """Returns raw response with installed plugins.

        ## Returns
        Raw response NetBox's installed plugins.

        ## Examples

        ```python
        nb.plugins.installed_plugins()
        [
            {
                'name': 'test_plugin',
                'package': 'test_plugin',
                'author': 'Dmitry',
                'description': 'Netbox test plugin',
                'verison': '0.10'
            }
        ]
        ```
        """
        installed_plugins = Request(
            base="{}/plugins/installed-plugins".format(
                self.api.base_url,
            ),
            token=self.api.token,
            http_session=self.api.http_session,
        ).get()
        return installed_plugins