# -*- coding: utf-8 -*-
# (c) Christian Meißner 2020
import json
import re
import requests
from requests.auth import HTTPBasicAuth
from phpypam.core.exceptions import PHPyPAMException
GET = requests.get
POST = requests.post
PATCH = requests.patch
DELETE = requests.delete
OPTIONS = requests.options
[docs]class Api(object):
""" The main class. It generates tha API object where you can run
different actions again to `create`, `update` and `delete` entities.
It also provides functions with informational character only.
"""
def __init__(self, url, app_id, username=None, password=None, token=None, encryption=False, timeout=None, ssl_verify=True, user_agent=None):
""" contructor method
:param url: The URL to a phpIPAM instance. It includes the protocol (`http` or `https`).
:type url: str
:param app_id: The app_id which is used for the API operations.
For modifying operations it needs read/write access.
:type app_id: str
:param username: The `username` which is used to connect to API., defaults to None
:type username: str, optional
:param password: The `password` to authenticate `username` against API., defaults to None
:type password: str, optional
:param token: A `token` if you want to use token based authentication., defaults to None
:type token: str, optional
:param encryption: Should request be encrypted. This is the needed encryption string., defaults to False
:type encryption: bool, optional
:param timeout: Seconds until a request will time out., defaults to None
:type timeout: int, optional
:param ssl_verify: Should certificate of endpoint verified or not.
Useful if you use a self signed certificate., defaults to True
:type ssl_verify: bool, optional
:param user_agent: With this parameter you can define a own user agent header string., defaults to None
:type user_agent: str, optional
"""
self._api_url = url
self._api_appid = app_id
self._api_username = username
self._api_password = password
self._api_token = token
self._api_encryption = encryption
self._api_timeout = timeout
self._api_ssl_verify = ssl_verify
self._api_headers = {
'content-type': 'application/json',
}
if user_agent:
self._api_headers['user-agent'] = user_agent
if not self._api_encryption:
self._login()
def _query(self, path='user', headers=None, method=GET, data=None, params=None, auth=None, token=None):
""" Sends queries to phpIPAM API in a generalistic manner
:param path: Path to the controler and possibly to function to use., defaults to 'user'
:type path: str, optional
:param headers: Optional request headers, defaults to None
:type headers: dict, optional
:param method: method to be used for the query (choice: 'GET', 'POST', 'PATCH', 'OPTIONS')., defaults to GET
:type method: requests method object, optional
:param data: Dictionary, list of tuples, bytes, or file-like object to send in the body of the :class:'Request'., defaults to None
:type data: dict, optional
:param params: Dictionary list of tuples or bytes to send in the query string for the :class:'Request'., defaults to None
:type params: dict, optional
:param auth: Auth tuple to enable Basic/Digest/Custom HTTP Auth., defaults to None
:type auth: HTTPBasicAuth object, optional
:type auth: str, optional
:param token: Api token get from last login., defaults to None
:type token: str, optional
:raises PHPyPAMException: [description]
:return: If query returns any data it returns this dict or list else it returns last exit status
:rtype: Union[bool, dict, list]
"""
_api_path = path
_api_headers = headers or {}
_method = method
_data = data
_params = params
_auth = auth
if self._api_token:
_api_headers['token'] = self._api_token
_url = '{}/api/{}/{}'.format(self._api_url, self._api_appid, _api_path)
if _params and not _url.endswith('/'):
_url = _url + '/'
resp = _method(
_url,
params=_params,
data=_data,
headers=_api_headers,
auth=_auth,
verify=self._api_ssl_verify,
timeout=self._api_timeout,
)
result = resp.json()
if result['code'] not in (200, 201) or not result['success']:
raise PHPyPAMException(code=result['code'], message=result['message'])
else:
if 'data' in result:
return result['data']
def _login(self):
""" Login method
"""
_auth = HTTPBasicAuth(self._api_username, self._api_password)
resp = self._query(method=POST, auth=_auth)
self._api_token = resp['token']
[docs] def get_token(self):
""" Method to grap last login token
:return: Returns the api token from the last successful login.
:rtype: str
"""
return self._api_token
[docs] def get_entity(self, controller, controller_path=None, params=None):
""" Method to get an existing entity
:param controller: Name of the controller to request entity from.
:type controller: str
:param controller_path: The path which is used to query for entities, defaults to None
:type controller_path: str, optional
:param params: Request parameters which have to be append to the request URI, defaults to None
:type params: dict, optional
:return: Result of the query. It can be either a 'list' or 'dict'.
:rtype: Union[dict, list]
"""
_path = controller
_controller_path = controller_path
_params = params
if _controller_path:
_path = '{}/{}'.format(_path, _controller_path)
return self._query(token=self._api_token, method=GET, path=_path, params=_params)
[docs] def create_entity(self, controller, controller_path=None, data=None, params=None):
""" Create an entity
:param controller: Name of the controller to use.
:type controller: str
:param controller_path: The path which is used to query for entities, defaults to None
:type controller_path: str, optional
:param data: Dictionary, list of tuples, bytes, or file-like object to send in the body of the :class:`Request`.
:type data: dict
:param params: Dictionary list of tuples or bytes to send in the query string for the :class:`Request`., defaults to None
:type params: dict, optional
:return: Returns the newly created entity.
:rtype: Union[dict, list]
"""
_path = controller
_controller_path = controller_path
_params = params
if _controller_path:
_path = '{}/{}'.format(_path, _controller_path)
return self._query(token=self._api_token, method=POST, path=_path, data=data, params=_params)
[docs] def delete_entity(self, controller, controller_path, params=None):
""" This method is used to delete an entity.
:param controller: Name of the controller to use.
:type controller: str
:param controller_path: The path wich is used to access the entity to delete.
:type controller_path: str
:param params: Dictionary, list of tuples or bytes to send in the query string for the :class:`Request`., defaults to None
:type params: dict, optional
:return: Returns True if entity was deleted successfully or either 'dict' or 'list' of entities to work on.
:rtype: Union[book, dict, list]
"""
_path = '{}/{}'.format(controller, controller_path)
_params = params
return self._query(token=self._api_token, method=DELETE, path=_path, params=_params)
[docs] def update_entity(self, controller, controller_path=None, data=None, params=None):
""" This method is used to update an entity.
:param controller: Name of the controller to use.
:type controller: str
:param controller_path: The path which is used to access the entity to update., defaults to None
:type controller_path: str, optional
:param data: Dictionary, list of tuples, bytes, or file-like object to send in the body of the :class:`Request`., defaults to None
:type data: dict, optional
:param params: Dictionary list of tuples or bytes to send in the query string for the :class:`Request`., defaults to None
:type params: dict, optional
:return: Returns either a 'dict' or 'list' of the changed entity
:rtype: Union[dict, list]
"""
_path = controller
_controller_path = controller_path
_params = params
if _controller_path:
_path = '{}/{}'.format(_path, _controller_path)
return self._query(token=self._api_token, method=PATCH, path=_path, data=data, params=_params)
[docs] def controllers(self):
""" This method is used to report all known controllers of phpIPAM API.
Unfortunately the API doesn't report all nor the correct paths for all 'controllers'.
:return: Returns a tuple of controller paths.
:rtype: tuple
"""
result = self._query(token=self._api_token, method=OPTIONS, path='/')
controllers = ({re.sub(r'^/api/' + self._api_appid + '/(.+)/$', r'\1', v) for ctrl in result['controllers'] for (k, v) in ctrl.items() if k == 'href'})
return controllers