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

Update refs to dict

Import Dict from typing
parent 733a97d0
Pipeline #74906 passed with stage
in 36 seconds
from typing import List
from typing import Dict, List
from lasloader.file_loader import IFileLoader, JsonLoader
......@@ -90,5 +90,5 @@ class Configuration:
"""
return self._none_propagating_get(self._none_propagating_get(self._config.get("data"), "default"), 'owners')
def _none_propagating_get(self, d: dict, key: str) -> any:
def _none_propagating_get(self, d: Dict, key: str) -> any:
return None if d is None else d.get(key)
import os
from pathlib import Path
from typing import List
from typing import Dict, List
import lasio
from lasio.las import LASFile
from knack.log import get_logger
......@@ -60,7 +60,7 @@ class JsonLoader:
"""
self._file_loader = file_loader
def load(self, path: str) -> dict[str, any]:
def load(self, path: str) -> Dict[str, any]:
"""
Load and parse a json file
:param path str : The path and filename of the file to load.
......
from typing import List, Union
from typing import Dict, List, Union
import httpx
import io
from knack.log import get_logger
......@@ -38,7 +38,7 @@ class OsduClient:
self._base_url = base_url
self._data_partition_id = data_partition_id
def _create_headers(self) -> dict[str, str]:
def _create_headers(self) -> Dict[str, str]:
"""
Create a new set of auth headers for OSDU
......@@ -179,11 +179,11 @@ class OsduClient:
else:
return []
def _send_request_json_response(self, method: str, url: str, content: any, json: dict[str, any], headers=None) -> any:
def _send_request_json_response(self, method: str, url: str, content: any, json: Dict[str, any], headers=None) -> any:
res = self._send_request(method, url, content, json, headers)
return res.json()
def _send_request(self, method: str, url: str, content: any, json: dict[str, any], headers=None) -> any:
def _send_request(self, method: str, url: str, content: any, json: Dict[str, any], headers=None) -> any:
logger.debug(f"{method}: {url}")
headers = self._create_headers() if headers is None else headers
......
from dataclasses import dataclass
from typing import List, Union, Optional
from typing import Dict, List, Union, Optional
import urllib
import lasio
from knack.log import get_logger
......@@ -14,16 +14,16 @@ logger = get_logger(__name__)
@dataclass
class Record:
kind: str
acl: dict[str, List[str]]
legal: dict[str, Union[List[str], str]]
data: dict[str, any]
acl: Dict[str, List[str]]
legal: Dict[str, Union[List[str], str]]
data: Dict[str, any]
id: Optional[str] = None
class WellboreRecord(Record):
_raw_data: dict[str, any]
_raw_data: Dict[str, any]
def __init__(self, data: dict[str, any]) -> None:
def __init__(self, data: Dict[str, any]) -> None:
self.kind = data.get("kind", None)
self.data = data.get("data", {})
self.acl = data.get("acl", {})
......@@ -32,11 +32,11 @@ class WellboreRecord(Record):
self._raw_data = data
def get_raw_data(self) -> dict[str, any]:
def get_raw_data(self) -> Dict[str, any]:
"""
Get the wellbore data used to construct this object
:returns: The wellbore data
:rtype: dict[str, any]
:rtype: Dict[str, any]
"""
return self._raw_data
......@@ -50,9 +50,9 @@ class WellboreRecord(Record):
class WellLogRecord(Record):
_raw_data: dict[str, any]
_raw_data: Dict[str, any]
def __init__(self, data: dict[str, any]) -> None:
def __init__(self, data: Dict[str, any]) -> None:
self.kind = data.get("kind", None)
self.data = data.get("data", {})
self.acl = data.get("acl", {})
......@@ -70,11 +70,11 @@ class WellLogRecord(Record):
curves = self.data['Curves'] if self.data and 'Curves' in self.data else []
return [c['CurveID'] for c in curves if 'CurveID' in c]
def get_raw_data(self) -> dict[str, any]:
def get_raw_data(self) -> Dict[str, any]:
"""
Get the well log data used to construct this object
:returns: The well log data
:rtype: dict[str, any]
:rtype: Dict[str, any]
"""
return self._raw_data
......@@ -90,13 +90,13 @@ class WellLogRecord(Record):
class AttributeBuilder:
# Common
def build_acl(self, config: Configuration) -> dict[str, List[str]]:
def build_acl(self, config: Configuration) -> Dict[str, List[str]]:
"""
Get the 'acl' (access control List) component of a record.
:param config Configuration: Record config object containing config variables.
:returns: The 'acl' component of a record.
:rtype: dict[str, List[str]]
:rtype: Dict[str, List[str]]
"""
if None in [config.data_default_viewers, config.data_default_owners]:
raise ValueError("Please ensure a data default viewers and owners have been provided in the config file.")
......@@ -106,13 +106,13 @@ class AttributeBuilder:
"owners": config.data_default_owners
}
def build_legal(self, config: Configuration) -> dict[str, Union[List[str], str]]:
def build_legal(self, config: Configuration) -> Dict[str, Union[List[str], str]]:
"""
Get the 'legal' component of a record.
:param config Configuration: Record config object containing config variables.
:returns: The 'legal' component of a record.
:rtype: dict[str, Union[List[str], str]]
:rtype: Dict[str, Union[List[str], str]]
"""
if None in [config.legal_tags, config.legal_relevant_data_countries, config.legal_status]:
raise ValueError(
......@@ -143,14 +143,14 @@ class AttributeBuilder:
# If the LAS data object does not have a 'UWI' attribute return None
return None
def _build_name_aliases(self, uwi: str, config: Configuration) -> List[dict[str, str]]:
def _build_name_aliases(self, uwi: str, config: Configuration) -> List[Dict[str, str]]:
"""
Determine what the NameAliases attribute should be, depending on the existence of a UWI.
:param uwi str: The UWI from the LASFile.
:param config Configuration: Record config object containing config variables.
:returns: The 'name aliases' component of a wellbore record's data attribute.
:rtype: List[dict[str, str]]
:rtype: List[Dict[str, str]]
"""
if uwi is not None:
if config.data_partition_id is None:
......@@ -164,13 +164,13 @@ class AttributeBuilder:
else:
return []
def build_wellbore_data(self, las: lasio.LASFile, config: Configuration) -> dict[str, any]:
def build_wellbore_data(self, las: lasio.LASFile, config: Configuration) -> Dict[str, any]:
"""
Get the 'data' component of a Wellbore record.
:param las LASFile: The LASFile object from which to build the 'data' component.
:returns: The 'data' component of a Wellbore record.
:rtype: dict[str, any]
:rtype: Dict[str, any]
"""
well_name = las.well.WELL.value
uwi = self._get_uwi(las)
......@@ -182,14 +182,14 @@ class AttributeBuilder:
}
# Well Log
def _build_curves(self, las: lasio.LASFile, data_partition_id: str) -> List[dict[str, str]]:
def _build_curves(self, las: lasio.LASFile, data_partition_id: str) -> List[Dict[str, str]]:
"""
Get the curves from a LAS data object.
:param las LASFile: The LASFile object from which to extract the curves.
:param str data_partition_id: The datapartition ID
:returns: The curves for the well log.
:rtype: List[dict[str, str]]
:rtype: List[Dict[str, str]]
"""
if data_partition_id is None:
raise ValueError("Please ensure a data partition ID has been provided in the config file.")
......@@ -212,14 +212,14 @@ class AttributeBuilder:
return []
return curves
def build_well_log_data(self, las: lasio.LASFile, config: Configuration, wellbore_id: str) -> dict[str, any]:
def build_well_log_data(self, las: lasio.LASFile, config: Configuration, wellbore_id: str) -> Dict[str, any]:
"""
Get the 'data' component of a Well Log record.
:param las LASFile: The LASFile object from which to build the 'data' component.
:param wellbore_id str: The wellbore_id object, often returned by OSDU.
:returns: The 'data' component of a Well Log record.
:rtype: dict[str, any]
:rtype: Dict[str, any]
"""
if None in [wellbore_id]:
......
import json
from typing import List, Tuple
from typing import Dict, List, Tuple
import urllib
from knack.log import get_logger
from lasio.las import LASFile
......@@ -54,14 +54,14 @@ class WellLogService:
welllog_record.data["Curves"] = self.recognize_curves_family(welllog_record.data.get("Curves"), data_partition_id)
return welllog_record
def recognize_curves_family(self, curves: List[dict[str, any]], data_partition_id: str) -> dict[str, any]:
def recognize_curves_family(self, curves: List[Dict[str, any]], data_partition_id: str) -> Dict[str, any]:
"""
For a set of curves call OSDU to get the curve family and
return an enriched set of curves. This makes no changes to data in OSDU.
:param List[dict[str, any]] curves: The curves to be enriched
:param List[Dict[str, any]] curves: The curves to be enriched
:param str data_partition_id: The data partition Id.
:return: the updated curves
:rtype: dict[str, any]
:rtype: Dict[str, any]
"""
if curves is None:
logger.warning("No curve data to recognize")
......
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