Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Commit

Permalink
Add APIContainter. (#134)
Browse files Browse the repository at this point in the history
* Add APIContainter

* Use DataSequence.

* Add additional parameter.

* Use .dataseq()

* Fix mypy.

* remove print
  • Loading branch information
kagrski authored Mar 1, 2023
1 parent f7cdb52 commit 9003c6e
Show file tree
Hide file tree
Showing 5 changed files with 44 additions and 80 deletions.
13 changes: 13 additions & 0 deletions vmngclient/api/api_containter.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
from __future__ import annotations

from typing import TYPE_CHECKING

from vmngclient.api.tenant_api import TenantsAPI

if TYPE_CHECKING:
from vmngclient.session import vManageSession


class APIContainter:
def __init__(self, session: vManageSession):
self.tenants = TenantsAPI(session)
11 changes: 7 additions & 4 deletions vmngclient/api/templates/payloads/tenant/tenant_model.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,16 +8,17 @@
from vmngclient.api.tenant_api import TenantsAPI
from vmngclient.dataclasses import TenantInfo, TierInfo
from vmngclient.session import vManageSession
from vmngclient.typed_list import DataSequence


@lru_cache
def get_tenants(session: vManageSession) -> List[TenantInfo]:
def get_tenants(session: vManageSession) -> DataSequence[TenantInfo]:
tenants_api = TenantsAPI(session)
return tenants_api.get_tenants()


@lru_cache
def get_tiers(session: vManageSession) -> List[TierInfo]:
def get_tiers(session: vManageSession) -> DataSequence[TierInfo]:
tenants_api = TenantsAPI(session)
return tenants_api.get_tiers()

Expand All @@ -41,8 +42,10 @@ def generate_payload(self, session: vManageSession) -> str:
tenants_api = TenantsAPI(session)

for tenant in self.tenants:
tier_info = tenants_api.get_tier(tenant.tier_name)
tenant_info = tenants_api.get_tenant(organization_name=tenant.organization_name)
tier_info = tenants_api.get_tiers().filter(name=tenant.tier_name).single_or_default()
tenant_info = (
tenants_api.get_tenants().filter(organization_name=tenant.organization_name).single_or_default()
)
# TODO Very, very ugly way...
tenant.__dict__["tier"] = tier_info
tenant.__dict__["info"] = tenant_info
Expand Down
90 changes: 15 additions & 75 deletions vmngclient/api/tenant_api.py
Original file line number Diff line number Diff line change
@@ -1,28 +1,20 @@
from typing import List, Optional
from __future__ import annotations

from vmngclient.api.basic_api import Device
from vmngclient.dataclasses import TenantInfo, TierInfo
from vmngclient.exceptions import InvalidOperationError
from vmngclient.session import vManageSession
from vmngclient.utils.creation_tools import create_dataclass
from typing import TYPE_CHECKING, Optional

from vmngclient.dataclasses import Device, TenantInfo, TierInfo

class TierNameNotFoundError(Exception):
pass
if TYPE_CHECKING:
from vmngclient.session import vManageSession

from vmngclient.typed_list import DataSequence

class TenantNameNotFoundError(Exception):
pass


# TODO tests
class TenantsAPI:
def __init__(self, session: vManageSession):
self.session = session

def get_tenants(
self, device_id: Optional[Device] = None, name: Optional[str] = None, organization_name: Optional[str] = None
) -> List[TenantInfo]:
def get_tenants(self, device_id: Optional[Device] = None) -> DataSequence[TenantInfo]:
"""Lists all the tenants on the vManage.
In a multitenant vManage system, this API is only avaiable in the Provider view.
Expand All @@ -31,71 +23,19 @@ def get_tenants(
device_id: Lists all tenants associated with a vSmart.
Returns:
List[TenantInfo]
DataSequence[TenantInfo]
"""

if device_id:
raise NotImplementedError()

response = self.session.get_data("/dataservice/tenant")
tenants = [create_dataclass(TenantInfo, tenant_info) for tenant_info in response]
if any([name, organization_name]) is False:
return tenants

# TODO
if name:
return list(filter(lambda tenant: tenant.name == name, tenants))

if organization_name:
return list(filter(lambda tenant: tenant.organization_name == organization_name, tenants))
return []

def get_tenant(self, name: Optional[str] = None, organization_name: Optional[str] = None) -> TenantInfo:
if any([name, organization_name]) is False:
raise ValueError("Argument `name` must be not null.")

tenants = self.get_tenants(name=name, organization_name=organization_name)
if not tenants:
raise TenantNameNotFoundError

if len(tenants) > 1:
raise InvalidOperationError("The input sequence contains more than one element.")

return tenants[0]

def get_tiers(self, name: Optional[str] = None) -> List[TierInfo]:
"""TODO"""

response = self.session.get_data("dataservice/device/tier")
tiers = [create_dataclass(TierInfo, tier) for tier in response]

if name is None:
return tiers

filtered_tiers = list(filter(lambda tier: tier.name == name, tiers))
return filtered_tiers

def get_tier(self, name: Optional[str] = None) -> TierInfo:
"""Gets tier with
Args:
name: Name of the tier. Defaults to None.
Raises:
TierNameNotFoundError: TODO
InvalidOperationError: Thrown when two or more objects
Returns:
TierInfo: _description_
"""
if name is None:
raise ValueError("Argument `name` must be not null.")
response = self.session.get("/dataservice/tenant")
tenants = response.dataseq(TenantInfo)

tiers = self.get_tiers(name)
if not tiers:
raise TierNameNotFoundError
return tenants

if len(tiers) > 1:
raise InvalidOperationError("The input sequence contains more than one element.")
def get_tiers(self) -> DataSequence[TierInfo]:
response = self.session.get(url="dataservice/device/tier")
tiers = response.dataseq(TierInfo)

return tiers[0]
return tiers
7 changes: 6 additions & 1 deletion vmngclient/dataclasses.py
Original file line number Diff line number Diff line change
Expand Up @@ -317,7 +317,10 @@ class TLOC:

@define(frozen=True)
class TierInfo(DataclassBase):
"""Endpoint: /dataservice/tier"""
"""Endpoint: /dataservice/tier
Since vManage 20.12 version, object has been renamed to "Resource Profile".
"""

name: str = field(metadata={FIELD_NAME: "tierName"})
vpn: int
Expand All @@ -329,6 +332,8 @@ class TierInfo(DataclassBase):
ipv6_route_limit_threshold: Optional[int] = field(default=None, metadata={FIELD_NAME: "ipv6RouteLimitThreshold"})
ipv6_route_limit: Optional[int] = field(default=None, metadata={FIELD_NAME: "ipv6RouteLimit"})
tlocs: List[TLOC] = field(factory=list)
# New in 20.12 version
nat_session_limit: Optional[int] = field(default=None, metadata={FIELD_NAME: "natSessionLimit"})


@define(frozen=True)
Expand Down
3 changes: 3 additions & 0 deletions vmngclient/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from requests.exceptions import ConnectionError, HTTPError, RequestException
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_fixed # type: ignore

from vmngclient.api.api_containter import APIContainter
from vmngclient.exceptions import InvalidOperationError
from vmngclient.response import response_history_debug, vManageResponse
from vmngclient.vmanage_auth import vManageAuth
Expand Down Expand Up @@ -169,6 +170,8 @@ def __init__(
super(vManageSession, self).__init__()
self.__prepare_session(verify, auth)

self.api = APIContainter(self)

def request(self, method, url, *args, **kwargs) -> vManageResponse:
full_url = self.get_full_url(url)
try:
Expand Down

0 comments on commit 9003c6e

Please sign in to comment.