### ===================================================================================================================
### MYVIIA base functionality
### ===================================================================================================================
# Copyright ©VIIA 2024
### ===================================================================================================================
### 1. Import modules
### ===================================================================================================================
# General imports
import os
from sys import platform
# Import ViiaSettings for server URL
from viiapackage.viiaSettings import ViiaSettings
# Module for requests library is used to connect to MYVIIA
import requests
from requests.exceptions import ConnectionError
# Collect the connection settings to connect to MYVIIA
if os.environ.get('USERNAME_MYVIIA') is None or os.environ.get('PASSWORD_MYVIIA') is None:
try:
from user_config import connection_dict
os.environ['USERNAME_MYVIIA'] = connection_dict[ViiaSettings.MYVIIA_SERVER]['email']
os.environ['PASSWORD_MYVIIA'] = connection_dict[ViiaSettings.MYVIIA_SERVER]['password']
use_user_config = True
except ImportError:
connection_dict = dict()
use_user_config = False
else:
use_user_config = True
### ===================================================================================================================
### 2. Collect viia_object information from MYVIIA database
### ===================================================================================================================
# Wrapper check function
[docs]def myviia_check_modules(func):
"""
This function checks if all required 3rd party modules are installed properly when
collecting data from MYVIIA webtool.
"""
def wrapper_function(*args, **kwargs):
if not use_user_config:
if 'project' in kwargs and not kwargs['project'].token:
raise ModuleNotFoundError(
"ERROR: The viiaPackage requires the user_config file to properly connect to the MYVIIA database. "
"Please make sure you have this file in your working directory.")
return func(*args, **kwargs)
return wrapper_function
# Wrapper check function
[docs]def myviia_token(func):
""" This function checks if token is provided or that it should be collected with login."""
def wrapper_function(*args, **kwargs):
if 'token' not in kwargs or kwargs['token'] is None:
kwargs['token'] = myviia_login()
return func(*args, **kwargs)
return wrapper_function
# Wrapper function for debug escape of posting
def myviia_debug(func):
""" This function checks if the post should be executed or that it should be logged only for debugging."""
def wrapper_function(*args, **kwargs):
if ViiaSettings.DEBUG:
return None
return func(*args, **kwargs)
return wrapper_function
def myviia_test_environment_available():
""" Checks if test environment of MYVIIA is available."""
if platform in ['linux', 'linux2']:
myviia_dev_available = os.getenv('MYVIIA_DEV_AVAILABLE')
if myviia_dev_available == 'True':
return True
print(f"MYVIIA test-environment is not available, tests will be skipped.")
return False
try:
from user_config import connection_dict
except ModuleNotFoundError:
return False
if 'myviia-test' in connection_dict:
return True
return False
### ===================================================================================================================
### 3. Request functions
### ===================================================================================================================
[docs]def _viia_get_from_myviia(url: str, token: str, stream: bool = False):
""" Function to create GET request to retrieve data from MYVIIA tool."""
if not stream:
response = requests.get(url=url, headers={'Authorization': f"Bearer {token}"})
if response.status_code >= 300:
raise ConnectionError(
f"ERROR: There is an issue with connecting to MYVIIA. Code: {response.status_code}. "
f"Reason: {response.reason}.")
return response
else:
return requests.get(url=url, headers={'Authorization': f"Bearer {token}"}, stream=True)
[docs]def _viia_post_to_myviia(url: str, token: str, data: dict):
""" Function to create POST request to enter new data record to MYVIIA tool."""
return requests.post(url=url, headers={'Authorization': f"Bearer {token}"}, json=data)
def _viia_patch_to_myviia(url: str, token: str, data: dict):
""" Function to create PATCH request to overwrite existing data record on MYVIIA tool."""
return requests.patch(url=url, headers={'Authorization': f"Bearer {token}"}, json=data)
def _viia_delete_from_myviia(url: str, token: str):
""" Function to create DELETE request to delete existing data record on MYVIIA tool."""
return requests.delete(url=url, headers={'Authorization': f"Bearer {token}"})
### ===================================================================================================================
### 4. Login to MYVIIA
### ===================================================================================================================
[docs]def myviia_login() -> str:
"""
This function connects to the MYVIIA database, provides authentication and returns a bearer token to acccess the
database for additional requests. Note that the token lifecycle is limited.
.. note:: Credentials should be stored in the user-config file. Do not save credentials in scripts.
Input:
- No input required.
Output:
- If successful authentication is performed, the token is returned as string.
- Error is raised if any problem occurs.
"""
# Collect the token from login
try:
login_request = requests.post(
url=f'{ViiaSettings.MYVIIA_SERVER_URL}/login',
data={'email': os.environ['USERNAME_MYVIIA'], 'password': os.environ['PASSWORD_MYVIIA']})
except ConnectionError:
raise ConnectionError("ERROR: Could not connect to the MYVIIA database. Please check your internet connection.")
except KeyError:
raise KeyError("ERROR: Could not retrieve the credentials from user_config file.")
# Check for denied access and collect the token that is required for further requests
if login_request.status_code == 401 and login_request.reason == 'Unauthorized':
raise PermissionError("ERROR: Access to the MYVIIA database is denied, check your credentials.")
elif login_request.status_code == 400 and login_request.reason == 'Bad Request':
for key in ['email', 'password']:
if key not in connection_dict['myviia']:
raise ValueError(
f"ERROR: Access to the MYVIIA database is denied, the '{key}' key is missing or incorrect.")
# Collect the token
login_request.raise_for_status()
login_request = login_request.json()
if 'token' not in login_request and 'message' in login_request['body'][0]:
raise PermissionError(
f"ERROR: API returns message: {login_request['body'][0]['message']}, "
f"please check input in your user_config.")
# Return the collected token
return login_request['token']
### ===================================================================================================================
### 5. End of script
### ===================================================================================================================