Merge pull request #135 from community-fabric/add_technology_models

Add technology models

.gitignore

@@ -83,6 +83,7 @@ ipython_config.py
 
 # pyenv
 .python-version
+pyvenv.cfg
 
 # pipenv
 #   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
@@ -109,6 +110,7 @@ venv/
 ENV/
 env.bak/
 venv.bak/
+bin/
 
 # Spyder project settings
 .spyderproject

CHANGELOG.md

@@ -1,6 +1,12 @@
 # Changelog
 
-## 5.0.2 (2022-07-13)
+## 5.0.3 (2022-08-11)
+### Fix
+* Deprecate IPF v4.X from package
+* Added Technologies to the SDK
+* Verification of API Version against IP Fabric Version
+
+## 5.0.2 (2022-08-09)
 ### Fix
 * Remove `importlib-metadata` dependency
 

ipfabric/api.py

@@ -3,9 +3,9 @@
 from typing import Optional
 from urllib.parse import urljoin
 
-import pkg_resources
 from httpx import Client
 from ipfabric_httpx_auth import PasswordCredentials, HeaderApiKey
+from pkg_resources import parse_version, get_distribution
 from pydantic import BaseSettings
 
 from ipfabric import models
@@ -52,38 +52,63 @@ def __init__(
         :param snapshot_id: str: IP Fabric snapshot ID to use by default for database actions - defaults to '$last'
         :param kwargs: dict: Keyword args to pass to httpx
         """
+        super().__init__()
+        self.headers.update({"Content-Type": "application/json"})
         with Settings() as settings:
-            self.api_version = api_version or settings.ipf_version
-            if not self.api_version:
-                ver = pkg_resources.get_distribution("ipfabric").version.split(".")
-                self.api_version = "v" + ver[0] + "." + ver[1]
-
             base_url = base_url or settings.ipf_url
-            if settings.ipf_dev:
-                kwargs["base_url"] = urljoin(base_url, f"{self.api_version}/")
-            else:
-                kwargs["base_url"] = urljoin(base_url, f"api/{self.api_version}/")
-            kwargs["verify"] = kwargs.get("verify") if "verify" in kwargs else settings.ipf_verify
+            if not base_url and not settings.ipf_url:
+                raise RuntimeError("IP Fabric base_url not provided or IPF_URL not set")
+
+            self.api_version, self.os_version = self.check_version(api_version or settings.ipf_version, base_url)
+            self.base_url = (
+                urljoin(base_url, f"api/{self.api_version}/")
+                if not settings.ipf_dev
+                else urljoin(base_url, f"{self.api_version}/")
+            )  # TODO: Verify 5.0 Dev Image stuff
+            self.verify = kwargs.get("verify") if "verify" in kwargs else settings.ipf_verify
             token = token or settings.ipf_token
             username = username or settings.ipf_username
             password = password or settings.ipf_password
 
-        if not kwargs["base_url"]:
-            raise RuntimeError("IP Fabric base_url not provided or IPF_URL not set")
         if not token and not (username and password):
             raise RuntimeError("IP Fabric Token or Username/Password not provided.")
 
-        super().__init__(**kwargs)
-        self.headers.update({"Content-Type": "application/json"})
         self.auth = HeaderApiKey(token) if token else PasswordCredentials(base_url, username, password)
 
-        # Request IP Fabric for the OS Version, by doing that we are also ensuring the token is valid
-        self.os_version = self.fetch_os_version()
+        # Get Snapshots, by doing that we are also ensuring the token is valid
         self.snapshots = self.get_snapshots()
         self.snapshot_id = snapshot_id
 
+    def check_version(self, api_version, base_url):
+        """
+        Checks API Version and returns the version to use in the URL and the OS Version
+        :param api_version: str: User defined API Version or None
+        :param base_url: str: URL of IP Fabric
+        :return: api_version, os_version
+        """
+        if api_version == "v1":
+            raise RuntimeError("IP Fabric Version < 5.0 support has been dropped, please use ipfabric==4.4.3")
+        dist_ver = get_distribution("ipfabric").version.split('.')
+        api_version = parse_version(api_version) if api_version else parse_version(f"{dist_ver[0]}.{dist_ver[1]}")
+
+        resp = self.get(urljoin(base_url, "api/version"), headers={"Content-Type": "application/json"})
+        resp.raise_for_status()
+        os_api_version = parse_version(resp.json()["apiVersion"])
+        if api_version > os_api_version:
+            logger.warning(
+                f"Specified API or SDK Version ({api_version}) is greater then "
+                f"OS API Version.\nUsing OS Version:  ({os_api_version})"
+            )
+            api_version = os_api_version
+        elif os_api_version.major > api_version.major:
+            raise RuntimeError(
+                f"OS Major Version {os_api_version.major} is greater then SDK Version "
+                f"{api_version.major}.  Please upgrade the Python SDK to the new major version."
+            )
+
+        return f"v{api_version.major}.{api_version.minor}", parse_version(resp.json()["releaseVersion"])
+
     def update(self):
-        self.os_version = self.fetch_os_version()
         self.snapshots = self.get_snapshots()
 
     @property
@@ -110,20 +135,6 @@ def snapshot_id(self, snapshot_id):
         else:
             self._snapshot_id = self.snapshots[snapshot_id].snapshot_id
 
-    def fetch_os_version(self):
-        """
-        Gets IP Fabric version to ensure token is correct
-        :return: str: IP Fabric version
-        """
-        res = self.get(url="os/version")
-        if not res.is_error:
-            try:
-                return pkg_resources.parse_version(res.json()["version"])
-            except KeyError as exc:
-                raise ConnectionError(f"Error While getting the OS version, no Version available, message: {exc.args}")
-        else:
-            raise ConnectionRefusedError("Verify URL and Token are correct.")
-
     def get_snapshots(self):
         """
         Gets all snapshots from IP Fabric and returns a dictionary of {ID: Snapshot_info}

ipfabric/client.py

@@ -3,9 +3,9 @@
 from typing import Optional, Union
 from urllib.parse import urlparse
 
-from ipfabric import models
 from ipfabric.api import IPFabricAPI
 from ipfabric.intent import Intent
+from ipfabric.models import Technology, Inventory
 
 DEFAULT_ID = "$last"
 
@@ -19,8 +19,9 @@ def wrapper(self, url, *args, **kwargs):
         if "filters" in kwargs and isinstance(kwargs["filters"], str):
             kwargs["filters"] = loads(kwargs["filters"])
         path = urlparse(url or kwargs["url"]).path
-        r = re.search(r"(api/)?v\d(\.\d)?", path)
-        url = path[r.end() + 1 :] if r else path
+        r = re.search(r"(api/)?v\d(\.\d)?/", path)
+        url = path[r.end():] if r else path
+        url = url[1:] if url[0] == '/' else url
         return func(self, url, *args, **kwargs)
 
     return wrapper
@@ -45,8 +46,9 @@ def __init__(
         :param kwargs: dict: Keyword args to pass to httpx
         """
         super().__init__(base_url, api_version, token, snapshot_id, username, password, **kwargs)
-        self.inventory = models.Inventory(client=self)
+        self.inventory = Inventory(client=self)
         self.intent = Intent(client=self)
+        self.technology = Technology(client=self)
 
     @check_format
     def fetch(
@@ -130,7 +132,7 @@ def fetch_all(
     @check_format
     def query(self, url: str, payload: Union[str, dict], all: bool = True):
         """
-        Submits a query, does no formating on the parameters.  Use for copy/pasting from the webpage.
+        Submits a query, does no formatting on the parameters.  Use for copy/pasting from the webpage.
         :param url: str: Example: https://demo1.ipfabric.io/api/v1/tables/vlan/device-summary
         :param payload: Union[str, dict]: Dictionary to submit in POST or can be JSON string (i.e. read from file).
         :param all: bool: Default use pager to get all results and ignore pagination information in the payload

ipfabric/models.py

@@ -4,6 +4,8 @@
 
 from pydantic import BaseModel, Field
 
+from ipfabric.technology import *
+
 logger = logging.getLogger()
 
 
@@ -169,8 +171,112 @@ def modules(self):
 
     @property
     def powerSupplies(self):
-        return Table(client=self.client, endpoint="tables/inventory/powerSupplies")
+        logger.warning(
+            """Use of client.inventory.PowerSupplies will be deprecated in a future release, please 
+                        use client.technology.platform_environment_power_supplies"""
+        )
+        return Table(client=self.client, endpoint="tables/inventory/power-supplies")
 
     @property
     def powerSuppliesFans(self):
-        return Table(client=self.client, endpoint="tables/inventory/powerSuppliesFans")
+        logger.warning(
+            """Use of client.inventory.PowerSuppliesFans will be deprecated in a future release, please 
+                        use client.technology.platform_environment_power_supplies_fans"""
+        )
+        return Table(client=self.client, endpoint="tables/inventory/power-supplies-fans")
+
+
+class Technology(BaseModel):
+    client: Any
+
+    @property
+    def platforms(self):
+        return Platforms(client=self.client)
+
+    @property
+    def interfaces(self):
+        return Interfaces(client=self.client)
+
+    @property
+    def neighbors(self):
+        return Neighbors(client=self.client)
+
+    @property
+    def dhcp(self):
+        return Dhcp(client=self.client)
+
+    @property
+    def port_channels(self):
+        return PortChannels(client=self.client)
+
+    @property
+    def vlans(self):
+        return Vlans(client=self.client)
+
+    @property
+    def stp(self):
+        return Stp(client=self.client)
+
+    @property
+    def addressing(self):
+        return Addressing(client=self.client)
+
+    @property
+    def fhrp(self):
+        return Fhrp(client=self.client)
+
+    @property
+    def managed_networks(self):
+        return ManagedNetworks(client=self.client)
+
+    @property
+    def mpls(self):
+        return Mpls(client=self.client)
+
+    @property
+    def multicast(self):
+        return Multicast(client=self.client)
+
+    @property
+    def cloud(self):
+        return Cloud(client=self.client)
+
+    @property
+    def management(self):
+        return Management(client=self.client)
+
+    @property
+    def ip_telephony(self):
+        return IpTelephony(client=self.client)
+
+    @property
+    def load_balancing(self):
+        return LoadBalancing(client=self.client)
+
+    @property
+    def oam(self):
+        return Oam(client=self.client)
+
+    @property
+    def qos(self):
+        return Qos(client=self.client)
+
+    @property
+    def routing(self):
+        return Routing(client=self.client)
+
+    @property
+    def sdn(self):
+        return Sdn(client=self.client)
+
+    @property
+    def sdwan(self):
+        return Sdwan(client=self.client)
+
+    @property
+    def security(self):
+        return Security(client=self.client)
+
+    @property
+    def wireless(self):
+        return Wireless(client=self.client)

ipfabric/technology/__init__.py

@@ -0,0 +1,49 @@
+from .addressing import Addressing
+from .cloud import Cloud
+from .dhcp import Dhcp
+from .fhrp import Fhrp
+from .interfaces import Interfaces
+from .ip_telephony import IpTelephony
+from .load_balancing import LoadBalancing
+from .managed_networks import ManagedNetworks
+from .management import Management
+from .mpls import Mpls
+from .multicast import Multicast
+from .neighbors import Neighbors
+from .oam import Oam
+from .platforms import Platforms
+from .port_channels import PortChannels
+from .qos import Qos
+from .routing import Routing
+from .sdn import Sdn
+from .sdwan import Sdwan
+from .security import Security
+from .stp import Stp
+from .vlans import Vlans
+from .wireless import Wireless
+
+__all__ = [
+    "Addressing",
+    "Dhcp",
+    "Fhrp",
+    "Interfaces",
+    "Management",
+    "ManagedNetworks",
+    "Mpls",
+    "Multicast",
+    "Neighbors",
+    "Platforms",
+    "PortChannels",
+    "Routing",
+    "Stp",
+    "Vlans",
+    "Cloud",
+    "IpTelephony",
+    "LoadBalancing",
+    "Oam",
+    "Qos",
+    "Sdn",
+    "Sdwan",
+    "Security",
+    "Wireless",
+]