-
Notifications
You must be signed in to change notification settings - Fork 9
/
operations.py
115 lines (87 loc) · 3.22 KB
/
operations.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
from typing import Dict
from vortexasdk.api.id import ID
from vortexasdk.client import default_client
from vortexasdk.exceptions import InvalidAPIDataResponseException
from vortexasdk.logger import get_logger
from vortexasdk.search_response import SearchResponse
from vortexasdk.utils import filter_exact_match
logger = get_logger(__name__)
class Reference:
"""Lookup Vortexa Reference Data using an entity ID."""
def __init__(self, resource):
"""
Init.
# Arguments
resource: The vortexa endpoint used for reference lookups.
"""
self._resource = resource
def reference(self, id: ID) -> Dict:
"""
Lookup reference data using ID.
# Arguments
id: ID of the entity we're looking up
# Returns
An entity matching the ID
# Examples
>>> Reference("/reference/geographies").reference(id='cfb8c4ef76585c3a37792b643791a0f4ff6d5656d5508927d8017319e21f2fca') # doctest: +SKIP
"""
logger.info(
f"Looking up {self.__class__.__name__} reference data with id: {id}"
)
data = default_client().get_reference(self._resource, id)
assert len(data) <= 1, InvalidAPIDataResponseException(
f"Server error: more than one record returned matching ID {id}"
)
try:
return data[0]
except IndexError:
return {}
class Search:
"""Search Vortexa Reference Data."""
def __init__(self, resource):
"""
Init.
# Arguments
resource: Appropriate search resource
"""
self._resource = resource
# This method has been renamed from `search` to `search_with_client` to avoid type signature
# issues with the `search` method in each endpoint class.
def search_with_client(
self,
exact_term_match: bool = None,
response_type: str = None,
headers: dict = None,
**api_params,
) -> SearchResponse:
"""
Search Reference data filtering on `params`.
# Arguments
exact_term_match: Optional argument to filter names on exact matches
api_params: Search parameters to be passed on to the API
response_type: Optional type of the response - to handle endpoints which do not support paging
# Returns
Result of VortexaAPI call from hitting querying the `resource` endpoint filtering with `params`.
# Examples
>>> Search("/reference/vessels").search(term="DHT") # doctest: +SKIP
"""
logger.info(f"Searching {self.__class__.__name__}")
api_result = default_client().search(
self._resource,
response_type=response_type,
headers=headers,
**api_params,
)
logger.debug(
f"{len(api_result)} results received from {self._resource}"
)
if exact_term_match:
logger.debug("Filtering results on exact term match")
return {
"reference": api_result["reference"],
"data": filter_exact_match(
api_params["term"], api_result["data"]
),
}
else:
return api_result