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

Use env vars for token and config path

parent d0139b44
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.osdu_client import OsduClient
from lasloader.configuration import Configuration
from lasloader.file_loader import LocalFileLoader
......@@ -10,19 +12,23 @@ from lasloader.well_service import WellLogService
logger = get_logger(__name__)
def download_las(token: str,
config_path: str,
welllog_id: str,
def download_las(welllog_id: str,
outfile: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH'),
curves: List[str] = None):
"""
Retrieve welllog data from an OSDU instance and save to a LAS format file
: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
:param str welllog_id: The well bore id of the record to retrieve
:param str outfile: The path of the output file
: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
: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
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.file_loader import LasParser, LocalFileLoader, FileUtilities
from lasloader.record_mapper import LasToRecordMapper
from lasloader.configuration import Configuration
......@@ -23,13 +25,16 @@ def printlas(input_path):
print(las_data.header)
def convert(input_path: str, config_path: str, wellbore_id: str):
def convert(input_path: str, wellbore_id: str, config_path: str = os.environ.get('CONFIGPATH')):
"""
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 config_path: Path to the LAS metadata file
: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
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.file_loader import LasParser, LocalFileLoader, FileUtilities
from lasloader.record_mapper import LasToRecordMapper
from lasloader.osdu_client import OsduClient
......@@ -14,8 +16,8 @@ logger = get_logger(__name__)
def wellbore(
input_path: str,
token: str,
config_path: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH'),
no_recognize: bool = False):
"""
Ingest a LAS file (single) or directory of LAS files (bulk) into OSDU
......@@ -24,6 +26,10 @@ 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
config = Configuration(LocalFileLoader(), config_path)
las_parser = LasParser(LocalFileLoader())
client = OsduClient(config.base_url, token, config.data_partition_id)
......@@ -57,8 +63,8 @@ def wellbore(
def welllog_data(
welllog_id: str,
input_path: str,
token: str,
config_path: str):
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH')):
"""
Write data from a LAS file to an existing Well Log.
:param str welllog_id: ID of well log to be updated.
......@@ -66,6 +72,10 @@ 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
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
service = WellLogService(client)
......
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.osdu_client import OsduClient
from lasloader.configuration import Configuration
from lasloader.file_loader import LocalFileLoader
......@@ -9,15 +11,19 @@ from lasloader.file_loader import LocalFileLoader
logger = get_logger(__name__)
def wellbore(token: str,
config_path: str,
wellbore_id: str):
def wellbore(wellbore_id: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH')):
"""
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
:param str wellbore_id: The well bore id of the record to retrieve
"""
if not validate_token_and_config_path(token, config_path):
logger.error("Listing failed: Process exiting")
return
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
logger.warning(f"Getting wellbore ID {wellbore_id}")
......@@ -26,16 +32,21 @@ def wellbore(token: str,
print(json.dumps(wellbore.get_raw_data(), indent=4, sort_keys=True))
def welllog(token: str,
config_path: str,
welllog_id: str,
def welllog(welllog_id: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH'),
curves: bool = False):
"""
Retrieve and print welllog 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
:param str wellbore_id: The well bore id of the record to retrieve
: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
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
logger.warning(f"Getting welllog ID {welllog_id}")
......@@ -48,17 +59,21 @@ def welllog(token: str,
print(json.dumps(welllog.get_raw_data(), indent=4, sort_keys=True))
def welllog_data(token: str,
config_path: str,
welllog_id: str,
def welllog_data(welllog_id: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH'),
curves: List[str] = None):
"""
Retrieve and print welllog data from an OSDU instance
:param str welllog_id: The welllog 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
:param str welllog_id: The welllog id of the record to retrieve
: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
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
logger.warning(f"Getting data for welllog ID {welllog_id}")
......
import json
import os
from knack.log import get_logger
from lasloader.command_utils import validate_token_and_config_path
from lasloader.osdu_client import OsduClient
from lasloader.configuration import Configuration
from lasloader.file_loader import LocalFileLoader
......@@ -9,15 +11,18 @@ from lasloader.well_service import WellBoreService, WellLogService
logger = get_logger(__name__)
def wellbore_search(token: str,
config_path: str,
wellbore_name: str):
def wellbore_search(wellbore_name: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH')):
"""
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
:param str wellbore_name: The well bore name to search for
"""
if not validate_token_and_config_path(token, config_path):
logger.error("Search failed: Process exiting")
return
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.configuration import Configuration
from lasloader.file_loader import LocalFileLoader
from lasloader.osdu_client import LasLoaderWebResponseError, OsduClient
......@@ -8,7 +10,10 @@ from lasloader.well_service import WellLogService
logger = get_logger(__name__)
def welllog(welllog_id: str, token: str, config_path: str, curve_families: bool = True):
def welllog(welllog_id: str,
token: str = os.environ.get('OSDUTOKEN'),
config_path: str = os.environ.get('CONFIGPATH'),
curve_families: bool = True):
"""
Update an existing well log record.
:param str welllog_id: ID of the well log to be updated.
......@@ -18,6 +23,10 @@ def welllog(welllog_id: str, token: str, config_path: str, curve_families: bool
"""
# 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
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(config.base_url, token, config.data_partition_id)
service = WellLogService(client)
......
Markdown is supported
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