Commit ba8fcb44 authored by Niall McDaid's avatar Niall McDaid
Browse files

Update method for env variables

parent 91d9f7d6
Pipeline #76699 passed with stage
in 1 minute and 26 seconds
from knack.log import get_logger
logger = get_logger(__name__)
def validate_token_and_config_path(token, config_path) -> bool:
"""
Validate that the JWT and config path variables have been set (i.e. they are not None).
If one or both of the values are None, log a relevant error message.
:param token: The token value to be validated.
:param config_path: The config path to be validated.
:return: True (if variables are not None) or False (if one/both are None).
"""
if token and config_path:
return True
if not token:
logger.error("A valid bearer token must be set as a command option (-t) OR as the environment variable 'OSDUTOKEN'.")
if not config_path:
logger.error("A path to a configuration file must be set as a command option (-c) OR as the environment variable 'CONFIGPATH'.")
return False
from typing import List
import os
from knack.log import get_logger
from lasloader.command_utils import validate_token_and_config_path
from lasloader.environment_var_helper import EnvironmentVarHelper
from lasloader.osdu_client import OsduClient
from lasloader.configuration import Configuration
from lasloader.file_loader import LocalFileLoader
......@@ -14,8 +12,8 @@ logger = get_logger(__name__)
def download_las(welllog_id: str,
outfile: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH'),
token: str = None,
config_path: str = None,
curves: List[str] = None):
"""
Retrieve welllog data from an OSDU instance and save to a LAS format file
......@@ -25,9 +23,8 @@ def download_las(welllog_id: str,
:param str config_path: Path to the las metadata file
:param List[str] curves: The curves to retrieve, use None to get all curves
"""
if not validate_token_and_config_path(token, config_path):
logger.error("Download failed: Process exiting")
return
token = EnvironmentVarHelper.get_token(token)
config_path = EnvironmentVarHelper.get_config_path(config_path)
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
......
import os
from pathlib import Path
from knack.log import get_logger
from lasloader.command_utils import validate_token_and_config_path
from lasloader.environment_var_helper import EnvironmentVarHelper
from lasloader.file_loader import LasParser, LocalFileLoader, FileUtilities
from lasloader.record_mapper import LasToRecordMapper
from lasloader.configuration import Configuration
......@@ -25,16 +24,14 @@ def printlas(input_path):
print(las_data.header)
def convert(input_path: str, wellbore_id: str, config_path: str = os.environ.get('CONFIGPATH')):
def convert(input_path: str, wellbore_id: str, config_path: str = None):
"""
Convert a LAS file to Wellbore and Well Log and write to json files.
:param str input_path: Path and filename of a LAS file or folder containing LAS files
:param str wellbore_id: The wellbore id
:param str config_path: Path to the LAS metadata file
"""
if not validate_token_and_config_path(token="", config_path=config_path):
logger.error("File conversion failed: Process exiting")
return
config_path = EnvironmentVarHelper.get_config_path(config_path)
las_parser = LasParser(LocalFileLoader())
config = Configuration(LocalFileLoader(), config_path)
......
import json
import os
from pathlib import Path
from ntpath import basename
from knack.log import get_logger
from lasloader.command_utils import validate_token_and_config_path
from lasloader.environment_var_helper import EnvironmentVarHelper
from lasloader.file_loader import LasParser, LocalFileLoader, FileUtilities
from lasloader.record_mapper import LasToRecordMapper
from lasloader.osdu_client import OsduClient
......@@ -16,8 +15,8 @@ logger = get_logger(__name__)
def wellbore(
input_path: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH'),
token: str = None,
config_path: str = None,
no_recognize: bool = False):
"""
Ingest a LAS file (single) or directory of LAS files (bulk) into OSDU
......@@ -26,9 +25,8 @@ def wellbore(
:param str config_path: Path to the LAS metadata file.
:param bool no_recognize: If true don't attempt to recognize the curves, otherwise recognize the curves
"""
if not validate_token_and_config_path(token, config_path):
logger.error("Ingest failed: Process exiting")
return
token = EnvironmentVarHelper.get_token(token)
config_path = EnvironmentVarHelper.get_config_path(config_path)
config = Configuration(LocalFileLoader(), config_path)
las_parser = LasParser(LocalFileLoader())
......@@ -63,8 +61,8 @@ def wellbore(
def welllog_data(
welllog_id: str,
input_path: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH')):
token: str = None,
config_path: str = None):
"""
Write data from a LAS file to an existing Well Log.
:param str welllog_id: ID of well log to be updated.
......@@ -72,9 +70,8 @@ def welllog_data(
:param str token: A valid bearer token that is used to authenticate the update request.
:param str config_path: Path to the LAS metadata file.
"""
if not validate_token_and_config_path(token, config_path):
logger.error("Ingest failed: Process exiting")
return
token = EnvironmentVarHelper.get_token(token)
config_path = EnvironmentVarHelper.get_config_path(config_path)
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
......
import json
from typing import List
import os
from knack.log import get_logger
from lasloader.command_utils import validate_token_and_config_path
from lasloader.environment_var_helper import EnvironmentVarHelper
from lasloader.osdu_client import OsduClient
from lasloader.configuration import Configuration
from lasloader.file_loader import LocalFileLoader
......@@ -12,17 +11,16 @@ logger = get_logger(__name__)
def wellbore(wellbore_id: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH')):
token: str = None,
config_path: str = None):
"""
Retrieve and print wellbore record from an OSDU instance
:param str wellbore_id: The well bore id of the record to retrieve
:param str token: a valid bearer token that is used to authenticate against the OSDU instance
:param str config_path: Path to the las metadata file
"""
if not validate_token_and_config_path(token, config_path):
logger.error("Listing failed: Process exiting")
return
token = EnvironmentVarHelper.get_token(token)
config_path = EnvironmentVarHelper.get_config_path(config_path)
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
......@@ -33,8 +31,8 @@ def wellbore(wellbore_id: str,
def welllog(welllog_id: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH'),
token: str = None,
config_path: str = None,
curves: bool = False):
"""
Retrieve and print welllog record from an OSDU instance
......@@ -43,9 +41,8 @@ def welllog(welllog_id: str,
:param str config_path: Path to the las metadata file
:param bool curves: Boolean to determine whether to list curves
"""
if not validate_token_and_config_path(token, config_path):
logger.error("Listing failed: Process exiting")
return
token = EnvironmentVarHelper.get_token(token)
config_path = EnvironmentVarHelper.get_config_path(config_path)
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
......@@ -60,8 +57,8 @@ def welllog(welllog_id: str,
def welllog_data(welllog_id: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH'),
token: str = None,
config_path: str = None,
curves: List[str] = None):
"""
Retrieve and print welllog data from an OSDU instance
......@@ -70,9 +67,8 @@ def welllog_data(welllog_id: str,
:param str config_path: Path to the las metadata file
:param List[str] curves: The curves to retrieve, use None to get all curves
"""
if not validate_token_and_config_path(token, config_path):
logger.error("Listing failed: Process exiting")
return
token = EnvironmentVarHelper.get_token(token)
config_path = EnvironmentVarHelper.get_config_path(config_path)
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
......
import json
import os
from knack.log import get_logger
from lasloader.command_utils import validate_token_and_config_path
from lasloader.environment_var_helper import EnvironmentVarHelper
from lasloader.osdu_client import OsduClient
from lasloader.configuration import Configuration
from lasloader.file_loader import LocalFileLoader
......@@ -12,17 +11,16 @@ logger = get_logger(__name__)
def wellbore_search(wellbore_name: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH')):
token: str = None,
config_path: str = None):
"""
Retrieve and print the ids of wellbores that match the specified name
:param str wellbore_name: The well bore name to search for
:param str token: a valid bearer token that is used to authenticate against the OSDU instance
:param str config_path: Path to the las metadata file
"""
if not validate_token_and_config_path(token, config_path):
logger.error("Search failed: Process exiting")
return
token = EnvironmentVarHelper.get_token(token)
config_path = EnvironmentVarHelper.get_config_path(config_path)
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
......
import os
from knack.log import get_logger
from lasloader.command_utils import validate_token_and_config_path
from lasloader.environment_var_helper import EnvironmentVarHelper
from lasloader.configuration import Configuration
from lasloader.file_loader import LocalFileLoader
from lasloader.osdu_client import LasLoaderWebResponseError, OsduClient
......@@ -11,8 +10,8 @@ logger = get_logger(__name__)
def welllog(welllog_id: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH'),
token: str = None,
config_path: str = None,
curve_families: bool = True):
"""
Update an existing well log record.
......@@ -23,9 +22,8 @@ def welllog(welllog_id: str,
"""
# This command currently only containes functionality to update the curve families of a well log.
# It is set up in such a way that it should be possible to extend to include other update operations in future if needed.
if not validate_token_and_config_path(token, config_path):
logger.error("Update failed: Process exiting")
return
token = EnvironmentVarHelper.get_token(token)
config_path = EnvironmentVarHelper.get_config_path(config_path)
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
......
import os
from knack.log import get_logger
logger = get_logger(__name__)
class MissingArgumentError(Exception):
"""
Custom error for missing arguments.
"""
def __init__(self, message="Required arguments not provided. See error logs for details."):
super().__init__(message)
class EnvironmentVarHelper:
def get_token(arg_token: str) -> str:
if arg_token is not None:
return arg_token
token = os.environ.get('OSDUTOKEN')
if token is not None:
return token
logger.error("A bearer token was not provided as an environment variable ('OSDUTOKEN') or command argument.")
raise MissingArgumentError
def get_config_path(arg_config_path: str) -> str:
if arg_config_path is not None:
return arg_config_path
config_path = os.environ.get('CONFIGPATH')
if config_path is not None:
return config_path
logger.error("A path to a configuration file was not provided as an environment variable ('CONFIGPATH') or command argument.")
raise MissingArgumentError
import pytest
from lasloader.command_utils import validate_token_and_config_path
def test_validate_token_and_config_path_returns_true_when_vars_not_None():
token = "an_example_token_string"
config_path = "path/to/config/file.json"
assert validate_token_and_config_path(token, config_path)
@pytest.mark.parametrize("token, config_path", [(None, None), (None, "path/to/file"), ("a token", None)])
def test_validate_token_and_config_path_returns_false_when_vars_None(token, config_path):
assert not validate_token_and_config_path(token, config_path)
import os
import pytest
from lasloader.environment_var_helper import EnvironmentVarHelper, MissingArgumentError
class TestGetToken:
def test_returns_original_token_if_not_None(self):
arg_token = "a token"
token = EnvironmentVarHelper.get_token(arg_token)
assert token == arg_token
def test_returns_environment_variable_if_original_token_is_None(self):
# Arrange
arg_token = None
env_var_token = "token stored as env variable"
os.environ['OSDUTOKEN'] = env_var_token
# Act
token = EnvironmentVarHelper.get_token(arg_token)
# Assert
assert token == env_var_token
# Teardown
os.environ.pop('OSDUTOKEN')
def test_raises_MissingArgumentError_when_original_token_is_None_and_environment_var_not_set(self):
arg_token = None
with pytest.raises(MissingArgumentError):
EnvironmentVarHelper.get_token(arg_token)
class TestGetConfigPath:
def test_returns_original_config_path_if_not_None(self):
arg_config_path = "path/to/config/file.json"
config_path = EnvironmentVarHelper.get_config_path(arg_config_path)
assert config_path == arg_config_path
def test_returns_environment_variable_if_original_config_path_is_None(self):
# Arrange
arg_config_path = None
env_var_config_path = "config/path/as/environment/variable.json"
os.environ['CONFIGPATH'] = env_var_config_path
# Act
config_path = EnvironmentVarHelper.get_config_path(arg_config_path)
# Assert
assert config_path == env_var_config_path
# Teardown
os.environ.pop('CONFIGPATH')
def test_raises_MissingArgumentError_when_original_config_path_is_None_and_environment_var_not_set(self):
arg_config_path = None
with pytest.raises(MissingArgumentError):
EnvironmentVarHelper.get_config_path(arg_config_path)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment