-
Notifications
You must be signed in to change notification settings - Fork 76
/
base.py
145 lines (115 loc) · 5.27 KB
/
base.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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
import zeep
import logging
from collections import OrderedDict
from netsuitesdk.internal.client import NetSuiteClient
from netsuitesdk.internal.utils import PaginatedSearch
from typing import List, Generator
logger = logging.getLogger(__name__)
# TODO: introduce arg and return types
class ApiBase:
def __init__(self, ns_client: NetSuiteClient, type_name):
self.ns_client = ns_client
self.type_name = type_name
def search(self, attribute, value, operator):
"""
Search Record
:param attribute: name of the field, eg. entityId
:param value: value of the field, eg. Amazon
:param operator: search matching operator, eg., 'contains', 'is', 'anyOf'
:return:
"""
records = self.ns_client.basic_stringfield_search(
type_name=self.type_name,
attribute=attribute,
value=value,
operator=operator
)
return records
def get_all(self):
generated_records = self.get_all_generator()
all_records = []
for records in generated_records:
all_records.extend(records)
return all_records
def count(self):
ps = PaginatedSearch(client=self.ns_client, type_name=self.type_name, pageSize=10, perform_search=True)
return ps.total_records
def get_all_generator(self, page_size=20):
"""
Returns a generator which is more efficient memory-wise
"""
return self.create_paginated_search(page_size=page_size)
def get(self, internalId=None, externalId=None) -> OrderedDict:
return self._get(internalId=internalId, externalId=externalId)
def get_ref(self, internalId=None, externalId=None) -> OrderedDict:
return self._serialize(self.ns_client.RecordRef(type=self.type_name.lower(),
internalId=internalId, externalId=externalId))
def post(self, data) -> OrderedDict:
raise NotImplementedError('post method not implemented')
def delete(self, internalId=None, externalId=None) -> OrderedDict:
record = self.ns_client.delete(recordType=self.type_name, internalId=internalId, externalId=externalId)
return record
def _serialize(self, record) -> OrderedDict:
"""
record: single record
Returns a dict
"""
return zeep.helpers.serialize_object(record)
def _serialize_array(self, records) -> List[OrderedDict]:
"""
records: a list of records
Returns an array of dicts
"""
return zeep.helpers.serialize_object(records)
@staticmethod
def _paginated_search_to_generator(paginated_search) -> List:
records = []
if paginated_search.num_records == 0:
return records
num_pages = paginated_search.total_pages
logger.debug('total pages = %d, records in page = %d', paginated_search.total_pages, paginated_search.num_records)
logger.debug(f'current page index {paginated_search.page_index}')
logger.debug('going to page %d', 0)
for p in range(1, num_pages + 1):
logger.debug('going to page %d', p)
paginated_search.goto_page(p)
logger.debug(f'current page index {paginated_search.page_index}')
records.extend(paginated_search.records)
return records
@staticmethod
def _paginated_search_generator(paginated_search: PaginatedSearch):
if paginated_search.num_records == 0:
return
num_pages = paginated_search.total_pages
logger.debug('total pages = %d, records in page = %d', paginated_search.total_pages, paginated_search.num_records)
logger.debug(f'current page index {paginated_search.page_index}')
logger.debug('going to page %d', 0)
for p in range(1, num_pages + 1):
logger.debug('going to page %d', p)
paginated_search.goto_page(p)
logger.debug(f'current page index {paginated_search.page_index}')
yield paginated_search.records
def create_paginated_search(self, page_size):
ps = PaginatedSearch(client=self.ns_client, type_name=self.type_name, pageSize=page_size)
return self._paginated_search_generator(paginated_search=ps)
def _search_all_generator(self, page_size):
ps = PaginatedSearch(client=self.ns_client, type_name=self.type_name, pageSize=page_size)
return self._paginated_search_to_generator(paginated_search=ps)
def _get_all(self) -> List[OrderedDict]:
records = self.ns_client.getAll(recordType=self.type_name)
return self._serialize_array(records)
def _get_all_generator(self):
res = self._get_all()
for r in res:
yield r
def _get(self, internalId=None, externalId=None) -> OrderedDict:
record = self.ns_client.get(recordType=self.type_name, internalId=internalId, externalId=externalId)
return record
def build_simple_fields(self, fields, source, target):
for field in fields:
if field in source:
target[field] = source[field]
def build_record_ref_fields(self, fields, source, target):
for field in fields:
if field in source and source[field]:
target[field] = self.ns_client.RecordRef(**(source[field]))