diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..c506968 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,6 @@ +{ + "python.pythonPath": "${workspaceFolder}/venv/bin/python", + "python.autoComplete.extraPaths": ["./tests/scripts"], + "python.envFile": "${workspaceFolder}/.env", + "restructuredtext.confPath": "${workspaceFolder}/docs" +} diff --git a/pyhpipam/__init__.py b/pyhpipam/__init__.py new file mode 100644 index 0000000..e4fcbcd --- /dev/null +++ b/pyhpipam/__init__.py @@ -0,0 +1,9 @@ +from pkg_resources import get_distribution, DistributionNotFound + +from pyhpipam.core.api import Api as api + + +try: + __version__ = get_distribution(__name__).version +except DistributionNotFound: + pass diff --git a/pyhpipam/core/__init__.py b/pyhpipam/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pyhpipam/core/api.py b/pyhpipam/core/api.py new file mode 100644 index 0000000..2349387 --- /dev/null +++ b/pyhpipam/core/api.py @@ -0,0 +1,58 @@ +# -*- coding: utf-8 -*- +# (c) Christian Meißner 2020 + +# pylint: disable=raise-missing-from +# pylint: disable=super-with-arguments + +import requests + +from requests.auth import HTTPBasicAuth +from pyhpipam.core.query import query + +GET = requests.get +POST = requests.post +PATCH = requests.patch +OPTIONS = requests.options + + +class Api(object): + + def __init__( + self, + url, + app_id, + username=None, + password=None, + token=None, + encryption=False, + timeout=None, + ssl_verify=True, + user_agent=None + ): + 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 _login(self): + _auth = HTTPBasicAuth(self._api_username, self._api_password) + resp = query(url=self._api_url, app_id=self._api_appid, method=POST, auth=_auth, verify=self._api_ssl_verify) + + self._token = resp['token'] + + def get_token(self): + return self._token diff --git a/pyhpipam/core/query.py b/pyhpipam/core/query.py new file mode 100644 index 0000000..4efc971 --- /dev/null +++ b/pyhpipam/core/query.py @@ -0,0 +1,81 @@ +# provides the main query methods. + +import json + + +class InvalidUsernameOrPasswordException(Exception): + def __init__(self, *args, **kwargs): + super(InvalidUsernameOrPasswordException, self).__init__(*args, **kwargs) + + +class EntityNotFoundException(Exception): + def __init__(self, *args, **kwargs): + super(EntityNotFoundException, self).__init__(*args, **kwargs) + + +class ParameterNotDefinedException(Exception): + def __init__(self, *args, **kwargs): + super(ParameterNotDefinedException, self).__init__(*args, **kwargs) + + +def query(**kwargs): + """ sends queries to phpIPAM API in a generalistic manner + + :param url: URL to the api of the phpIPAM instace. Needs to include the used scheme like ```https://``` or ```http://```. + :param path: Path to the controler and possibly to function to use. Default is ```user```. + :param app_id: Name of the app id used for this request. + :param method: method to be used for the query (choice: ```GET```, ```POST```, ```PATCH```, ```OPTIONS```). + :param headers: Headers for request. + :param timeout: Timeout for request. + :param verify: Should ssl certificate be verified. Default ```False```. + :param data: (optional) Dictionary Dictionary, list of tuples, bytes, or file-like object to send in the body of the :class:`Request`. + :param params: (optional) Dictionary list of tuples or bytes to send in the query string for the :class:`Request`. + :param auth: (optional) Auth tuple to enable Basic/Digest/Custom HTTP Auth. + :param token: (optional) Api token get from last login. + """ + + _api_url = kwargs.pop('url', None) + _api_path = kwargs.pop('path', 'user') + _api_appid = kwargs.pop('app_id', None) + _api_headers = kwargs.pop('headers', {}) + _method = kwargs.pop('method', 'GET') + _data = kwargs.pop('data', None) + _params = kwargs.pop('params', {}) + _auth = kwargs.pop('auth', None) + _api_token = kwargs.pop('token', None) + _api_timeout = kwargs.pop('timeout', None) + _api_ssl_verify = kwargs.pop('verify', False) + + if _api_url is None: + raise ParameterNotDefinedException('Parameter `url` not defined.') + + if _api_appid is None: + raise ParameterNotDefinedException('Parameter `app_id` not defined.') + + if _api_token: + _api_headers['token'] = _api_token + + _url = '{}/api/{}/{}'.format(_api_url, _api_appid, _api_path) + + if _data is not None: + _data = json.dumps(_data) + + resp = _method( + _url, + params=_params, + data=_data, + headers=_api_headers, + auth=_auth, + verify=_api_ssl_verify, + timeout=_api_timeout, + ) + + result = resp.json() + + if result['code'] not in (200, 201) or not result['success']: + if result['code'] == 500: + if result['message'] == 'Invalid username or password': + raise InvalidUsernameOrPasswordException(result['message']) + else: + if 'data' in result: + return result['data']