Skip to content

Add type hints in the code base #81

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -100,3 +100,6 @@ ENV/

# Visual Studio Code settings
.vscode/

# Local testing
pyixapi/__main__.py
16 changes: 7 additions & 9 deletions docs/conf.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# -*- coding: utf-8 -*-
# noqa: INP001
#
# pyixapi documentation build configuration file, created by
# sphinx-quickstart on Sun Sep 18 18:31:27 2022.
Expand All @@ -16,12 +16,12 @@
# add these directories to sys.path here. If the directory is relative to the
# documentation root, use os.path.abspath to make it absolute, like shown here.
#
import os
import sys
from pathlib import Path

from pkg_resources import get_distribution
from pyixapi import __version__ as pyixapi_version

sys.path.insert(0, os.path.abspath("../"))
sys.path.insert(0, str(Path().parent.resolve()))


# -- General configuration ------------------------------------------------
Expand Down Expand Up @@ -49,14 +49,14 @@

# General information about the project.
project = "pyixapi"
copyright = "2012, Guillaume Mazoyer"
copyright = "2022, Guillaume Mazoyer"
author = "Guillaume Mazoyer"

# The version info for the project you're documenting, acts as replacement for
# |version| and |release|, also used in various other places throughout the
# built documents.
# The full version, including alpha/beta/rc tags.
release = get_distribution("pyixapi").version
release = pyixapi_version
#
# The short X.Y version.
version = ".".join(release.split(".")[:2])
Expand Down Expand Up @@ -98,9 +98,7 @@
# so a file named "default.css" will overwrite the builtin "default.css".
# html_static_path = ['_static']

html_sidebars = {
"**": ["globaltoc.html", "relations.html", "sourcelink.html", "searchbox.html"]
}
html_sidebars = {"**": ["globaltoc.html", "relations.html", "sourcelink.html", "searchbox.html"]}

# -- Options for HTMLHelp output ------------------------------------------

Expand Down
2 changes: 1 addition & 1 deletion pyixapi/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from .core.api import API as api
from .core.api import API as api # noqa: N811
from .core.api import __version__ # noqa: F401
from .core.query import ContentError, RequestError

Expand Down
68 changes: 27 additions & 41 deletions pyixapi/core/api.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from typing import Any

import requests

from pyixapi.core.endpoint import Endpoint
Expand Down Expand Up @@ -32,7 +34,7 @@
__version__ = "0.2.2"


class API(object):
class API:
"""
The API object is the entrypoint for pyixapi.

Expand All @@ -42,13 +44,13 @@ class API(object):

def __init__(
self,
url,
key,
secret,
access_token="",
refresh_token="",
user_agent=f"pyixapi/{__version__}",
):
url: str,
key: str,
secret: str,
access_token: str = "",
refresh_token: str = "",
user_agent: str = f"pyixapi/{__version__}",
) -> None:
self.url = url.rstrip("/")
self.key = key
self.secret = secret
Expand All @@ -65,54 +67,40 @@ def __init__(
self.facilities = Endpoint(self, "facilities", model=Facility)
self.ips = Endpoint(self, "ips", model=IP)
self.macs = Endpoint(self, "macs", model=MAC)
self.network_feature_configs = Endpoint(
self, "network-feature-configs", model=NetworkFeatureConfig
)
self.network_feature_configs = Endpoint(self, "network-feature-configs", model=NetworkFeatureConfig)
self.network_features = Endpoint(self, "network-features", model=NetworkFeature)
self.network_service_configs = Endpoint(
self, "network-service-configs", model=NetworkServiceConfig
)
self.network_service_configs = Endpoint(self, "network-service-configs", model=NetworkServiceConfig)
self.network_services = Endpoint(self, "network-services", model=NetworkService)
self.pops = Endpoint(self, "pops", model=PoP)
# Version 2+
self.member_joining_rules = Endpoint(
self, "member-joining-rules", model=MemberJoiningRule
)
self.member_joining_rules = Endpoint(self, "member-joining-rules", model=MemberJoiningRule)
self.metro_areas = Endpoint(self, "metro-areas", model=MetroArea)
self.metro_area_networks = Endpoint(
self, "metro-area-networks", model=MetroAreaNetwork
)
self.metro_area_networks = Endpoint(self, "metro-area-networks", model=MetroAreaNetwork)
self.ports = Endpoint(self, "ports", model=Port)
self.port_reservations = Endpoint(
self, "port-reservations", model=PortReservation
)
self.port_reservations = Endpoint(self, "port-reservations", model=PortReservation)
self.roles = Endpoint(self, "roles", model=Role)
self.role_assignments = Endpoint(self, "role-assignments", model=RoleAssignment)

@property
def version(self):
def version(self) -> int:
"""
Get the API version of IX-API.
"""
return Request(
base=self.url, token=self.access_token, http_session=self.http_session
).get_version()
return Request(base=self.url, token=self.access_token, http_session=self.http_session).get_version()

@property
def accounts(self):
return Endpoint(
self, "customers" if self.version == 1 else "accounts", model=Account
)
def accounts(self) -> Endpoint:
return Endpoint(self, "customers" if self.version == 1 else "accounts", model=Account)

@property
def product_offerings(self):
def product_offerings(self) -> Endpoint:
return Endpoint(
self,
"products" if self.version == 1 else "product-offerings",
model=ProductOffering,
)

def authenticate(self):
def authenticate(self) -> Record:
"""
Authenticate and generate a pair of tokens.

Expand All @@ -130,16 +118,16 @@ def authenticate(self):
if self.refresh_token and not self.refresh_token.is_expired:
return self.refresh_authentication()

r = Request(
cat(self.url, "auth", "token"), http_session=self.http_session
).post(data={"api_key": self.key, "api_secret": self.secret})
r = Request(cat(self.url, "auth", "token"), http_session=self.http_session).post(
data={"api_key": self.key, "api_secret": self.secret}
)

self.access_token = Token.from_jwt(r["access_token"])
self.refresh_token = Token.from_jwt(r["refresh_token"])

return Record(r, self, self.auth)

def refresh_authentication(self):
def refresh_authentication(self) -> Record:
"""
Prolong authentication by refreshing the tokens pair.
"""
Expand All @@ -154,7 +142,7 @@ def refresh_authentication(self):

return Record(r, self, self.auth)

def health(self):
def health(self) -> dict[str, Any]:
"""
Get the health information from IX-API.

Expand All @@ -163,6 +151,4 @@ def health(self):
if self.version == 1:
return {}

return Request(
base=self.url, token=self.access_token, http_session=self.http_session
).get_health()
return Request(base=self.url, token=self.access_token, http_session=self.http_session).get_health()
26 changes: 16 additions & 10 deletions pyixapi/core/endpoint.py
Original file line number Diff line number Diff line change
@@ -1,26 +1,32 @@
from http import HTTPStatus
from typing import TYPE_CHECKING

from pyixapi.core.query import Request, RequestError
from pyixapi.core.response import Record, RecordSet
from pyixapi.core.util import cat

if TYPE_CHECKING:
from .api import API


class Endpoint(object):
class Endpoint:
"""
Represent actions available on endpoints in the IX-API.

Build the correct URL to make queries and the proper :py:class:`.Response`
object.
"""

def __init__(self, api, name, model=None):
def __init__(self, api: "API", name: str, model: Record | None = None) -> None:
self.return_obj = model if model else Record
self.api = api
self.url = cat(api.url, name)
self.name = name

def __str__(self):
def __str__(self) -> str:
return self.url

def all(self):
def all(self) -> RecordSet:
"""
Return all objects from an endpoint.
"""
Expand All @@ -31,7 +37,7 @@ def all(self):
)
return RecordSet(self, r)

def filter(self, *args, **kwargs):
def filter(self, *args, **kwargs) -> RecordSet:
"""
Query the list of a given endpoint. Also take named arguments that match the
usable filters on a given endpoint.
Expand All @@ -44,7 +50,7 @@ def filter(self, *args, **kwargs):
)
return RecordSet(self, r)

def get(self, *args, **kwargs):
def get(self, *args, **kwargs) -> Record | None:
"""
Return a single object from an endpoint.
"""
Expand Down Expand Up @@ -78,12 +84,12 @@ def get(self, *args, **kwargs):
try:
return next(RecordSet(self, r), None)
except RequestError as e:
if e.req.status_code == 404:
if e.req.status_code == HTTPStatus.NOT_FOUND:
return None
else:
raise e

def create(self, *args, **kwargs):
raise e

def create(self, *args, **kwargs) -> Record:
"""
Creates an object on an endpoint.

Expand Down
Loading