a.garaev/isp-maintenance
1
0
Fork 0
forked from ISPsystem/isp-maintenance
Code Pull Requests Activity

Refactoring: move API init statments to settings

Browse Source
This commit is contained in:
Stepan Zhukovsky 2024-06-06 00:08:10 +09:00
parent 5405679640
commit 8ad5e2a230
3 changed files with 50 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
mgrctl/api/base.py
View File

@ -3,52 +3,25 @@ import json
import urllib import urllib
import requests import requests
from mgrctl.settings.api import INPUT_URL, HEADERS from mgrctl.settings.api import (
from mgrctl.settings.platform import ( API_URL,
PLATFORM_TYPE, API_HEADERS,
PLATFORM_VERIFY_SSL, API_EMAIL,
PLATFORM_DUMMY, API_PASSWORD,
PLATFORM_DUMMY_VM6_API_URL, API_VERIFY_SSL
PLATFORM_DUMMY_VM6_EMAIL,
PLATFORM_DUMMY_VM6_PASSWORD,
PLATFORM_DUMMY_VM6_TOKEN,
PLATFORM_DUMMY_DCI6_API_URL,
PLATFORM_DUMMY_DCI6_EMAIL,
PLATFORM_DUMMY_DCI6_PASSWORD,
PLATFORM_DUMMY_DCI6_TOKEN
) )
class BaseAPI(object): class BaseAPI(object):
def __init__(self): def __init__(self):
"""Announces required parameters""" """Announces required parameters"""
if PLATFORM_TYPE == 'vm': self.API_URL = API_URL
if PLATFORM_DUMMY: self.API_HEADERS = API_HEADERS
self.API_URL = PLATFORM_DUMMY_VM6_API_URL self.API_EMAIL = API_EMAIL
self.AUTH_TYPE = 'Public' self.API_PASSWORD = API_PASSWORD
self.HEADERS = {'x-xsrf-token': PLATFORM_DUMMY_VM6_TOKEN} self.API_VERIFY_SSL = API_VERIFY_SSL
self.EMAIL = PLATFORM_DUMMY_VM6_EMAIL
self.PASSWORD = PLATFORM_DUMMY_VM6_PASSWORD
else:
self.API_URL = INPUT_URL
self.AUTH_TYPE = 'Internal'
self.HEADERS = HEADERS
if PLATFORM_TYPE == 'dci':
if PLATFORM_DUMMY:
self.API_URL = PLATFORM_DUMMY_DCI6_API_URL
self.AUTH_TYPE = 'Public'
self.HEADERS = {'x-xsrf-token': PLATFORM_DUMMY_DCI6_TOKEN}
self.EMAIL = PLATFORM_DUMMY_DCI6_EMAIL
self.PASSWORD = PLATFORM_DUMMY_DCI6_PASSWORD
else:
self.API_URL = INPUT_URL
self.AUTH_TYPE = 'Internal'
self.HEADERS = HEADERS
self.API_VERSION = 'v3' self.API_VERSION = 'v3'
self.API_DEFINITION = 'api' self.API_DEFINITION = 'api'
self.VERIFY_SSL = PLATFORM_VERIFY_SSL
def _gen_request_url(self, url): def _gen_request_url(self, url):
return f'{self.API_URL}/{self.API_DEFINITION}/{self.API_VERSION}{url}' return f'{self.API_URL}/{self.API_DEFINITION}/{self.API_VERSION}{url}'
@ -58,21 +31,21 @@ class BaseAPI(object):
with requests.Session() as session: with requests.Session() as session:
try: try:
url = self._gen_request_url(url) url = self._gen_request_url(url)
headers = self.HEADERS if not headers else headers headers = self.API_HEADERS if not headers else headers
params_str = urllib.parse.urlencode(data, safe="+'()") params_str = urllib.parse.urlencode(data, safe="+'()")
if method == 'POST': if method == 'POST':
api_request = session.post( api_request = session.post(
url=url, url=url,
json=data, json=data,
headers=headers, headers=headers,
verify=self.VERIFY_SSL verify=self.API_VERIFY_SSL
) )
if method == 'GET': if method == 'GET':
url = f'{url}?{params_str}' if params_str else url url = f'{url}?{params_str}' if params_str else url
api_request = session.get( api_request = session.get(
url=url, url=url,
headers=headers, headers=headers,
verify=self.VERIFY_SSL verify=self.API_VERIFY_SSL
) )
except Exception as error: except Exception as error:
api_request = { api_request = {
@ -103,8 +76,8 @@ class BaseAuthAPI(BaseAPI):
self.API_DEFINITION = 'auth' self.API_DEFINITION = 'auth'
def get_auth_token(self, email=None, password=None) -> dict: def get_auth_token(self, email=None, password=None) -> dict:
email = self.EMAIL if not email else email email = self.API_EMAIL if not email else email
password = self.PASSWORD if not password else password password = self.API_PASSWORD if not password else password
return self.call_api( return self.call_api(
url='/public/token', url='/public/token',
method='POST', method='POST',
@ -113,7 +86,7 @@ class BaseAuthAPI(BaseAPI):
def get_auth_key(self, token=None, user=None) -> dict: def get_auth_key(self, token=None, user=None) -> dict:
headers = {} headers = {}
user = self.EMAIL if not user else user user = self.API_EMAIL if not user else user
if token: if token:
headers = self.make_auth_header(token) headers = self.make_auth_header(token)
return self.call_api( return self.call_api(

27
mgrctl/settings/api.py
View File

@ -1,20 +1,30 @@
from requests.packages import urllib3 from requests.packages import urllib3
from mgrctl.settings.platform import ( from mgrctl.settings.platform import (
PLATFORM_TYPE, PLATFORM_TYPE,
PLATFORM_VERIFY_SSL_WARNING PLATFORM_VERIFY_SSL,
PLATFORM_VERIFY_SSL_WARNING,
PLATFORM_DUMMY,
PLATFORM_DUMMY_API_URL,
PLATFORM_DUMMY_EMAIL,
PLATFORM_DUMMY_PASSWORD,
PLATFORM_DUMMY_TOKEN,
) )
# Name of nginx container: # Name of nginx container:
INPUT_HOSTNAME = 'input' if PLATFORM_TYPE == 'vm' else 'dci_input_1' API_INPUT_HOSTNAME = 'input' if PLATFORM_TYPE == 'vm' else 'dci_input_1'
# Port that nginx container is listening: # Port that nginx container is listening:
INPUT_PORT = '1500' API_INPUT_PORT = '1500'
# Internal API url: # Internal API url:
INPUT_URL = f'http://{INPUT_HOSTNAME}:{INPUT_PORT}' API_URL = f'http://{API_INPUT_HOSTNAME}:{API_INPUT_PORT}'
# Headers for internal auth: # Headers for internal auth:
HEADERS = {"Internal-Auth": "on", "Accept": "application/json"} API_HEADERS = {"Internal-Auth": "on", "Accept": "application/json"}
# Alias for import:
API_VERIFY_SSL = PLATFORM_VERIFY_SSL
# Suppress warning from urllib3: # Suppress warning from urllib3:
if not PLATFORM_VERIFY_SSL_WARNING: if not PLATFORM_VERIFY_SSL_WARNING:
@ -23,3 +33,10 @@ if not PLATFORM_VERIFY_SSL_WARNING:
urllib3.disable_warnings( urllib3.disable_warnings(
category=urllib3.exceptions.InsecureRequestWarning category=urllib3.exceptions.InsecureRequestWarning
) )
# Development mode:
if PLATFORM_DUMMY:
API_URL = PLATFORM_DUMMY_API_URL
API_HEADERS = {'x-xsrf-token': PLATFORM_DUMMY_TOKEN}
API_EMAIL = PLATFORM_DUMMY_EMAIL
API_PASSWORD = PLATFORM_DUMMY_PASSWORD

20
mgrctl/settings/platform.py
View File

@ -22,12 +22,14 @@ PLATFORM_URL = env.str(
# Development mode: # Development mode:
PLATFORM_DUMMY = env.bool('PLATFORM_DUMMY', False) PLATFORM_DUMMY = env.bool('PLATFORM_DUMMY', False)
PLATFORM_DUMMY_VM6_API_URL = env.str('PLATFORM_DUMMY_VM6_API_URL', '') if PLATFORM_DUMMY:
PLATFORM_DUMMY_VM6_EMAIL = env.str('PLATFORM_DUMMY_VM6_EMAIL', '') if PLATFORM_TYPE == 'vm':
PLATFORM_DUMMY_VM6_PASSWORD = env.str('PLATFORM_DUMMY_VM6_PASSWORD', '') PLATFORM_DUMMY_API_URL = env.str('PLATFORM_DUMMY_VM6_API_URL', '')
PLATFORM_DUMMY_VM6_TOKEN = env.str('PLATFORM_DUMMY_VM6_TOKEN', '') PLATFORM_DUMMY_EMAIL = env.str('PLATFORM_DUMMY_VM6_EMAIL', '')
PLATFORM_DUMMY_PASSWORD = env.str('PLATFORM_DUMMY_VM6_PASSWORD', '')
PLATFORM_DUMMY_DCI6_API_URL = env.str('PLATFORM_DUMMY_DCI6_API_URL', '') PLATFORM_DUMMY_TOKEN = env.str('PLATFORM_DUMMY_VM6_TOKEN', '')
PLATFORM_DUMMY_DCI6_EMAIL = env.str('PLATFORM_DUMMY_DCI6_EMAIL', '') if PLATFORM_TYPE == 'dci':
PLATFORM_DUMMY_DCI6_PASSWORD = env.str('PLATFORM_DUMMY_DCI6_PASSWORD', '') PLATFORM_DUMMY_API_URL = env.str('PLATFORM_DUMMY_DCI6_API_URL', '')
PLATFORM_DUMMY_DCI6_TOKEN = env.str('PLATFORM_DUMMY_DCI6_TOKEN', '') PLATFORM_DUMMY_EMAIL = env.str('PLATFORM_DUMMY_DCI6_EMAIL', '')
PLATFORM_DUMMY_PASSWORD = env.str('PLATFORM_DUMMY_DCI6_PASSWORD', '')
PLATFORM_DUMMY_TOKEN = env.str('PLATFORM_DUMMY_DCI6_TOKEN', '')