Flesh out EDA client a bit more
This commit is contained in:
parent
e618057337
commit
c917605df9
@ -1,3 +1,10 @@
|
|||||||
|
from csv import DictReader
|
||||||
|
import xml.etree.ElementTree as ET
|
||||||
|
|
||||||
|
import requests
|
||||||
|
from requests.auth import HTTPBasicAuth
|
||||||
|
|
||||||
|
|
||||||
class EDAClientBase(object):
|
class EDAClientBase(object):
|
||||||
def list_contracts(
|
def list_contracts(
|
||||||
self,
|
self,
|
||||||
@ -97,8 +104,24 @@ class MockEDAClient(EDAClientBase):
|
|||||||
|
|
||||||
|
|
||||||
class EDAClient(EDAClientBase):
|
class EDAClient(EDAClientBase):
|
||||||
def __init__(self, base_url, user_name, user_role):
|
def __init__(self, base_url, user_name, user_role, auth_name, auth_pass):
|
||||||
pass
|
self.base_url = base_url
|
||||||
|
self.user_name = user_name
|
||||||
|
self.user_role = user_role
|
||||||
|
self.auth = HTTPBasicAuth(auth_name, auth_pass)
|
||||||
|
|
||||||
|
def _make_url(self, method, **kwargs):
|
||||||
|
query_args = dict(kwargs)
|
||||||
|
query_string = "&".join(
|
||||||
|
["{}={}".format(key, value) for key, value in query_args.items()]
|
||||||
|
)
|
||||||
|
return "{base_url}/{method}?{query_string}".format(
|
||||||
|
base_url=self.base_url, method=method, query_string=query_string
|
||||||
|
)
|
||||||
|
|
||||||
|
def _get(self, method, **kwargs):
|
||||||
|
url = self._make_url(method, **kwargs)
|
||||||
|
return requests.get(url, auth=self.auth, verify="ssl/server-certs/eda.pem")
|
||||||
|
|
||||||
def list_contracts(
|
def list_contracts(
|
||||||
self,
|
self,
|
||||||
@ -107,11 +130,37 @@ class EDAClient(EDAClientBase):
|
|||||||
cage_code=None,
|
cage_code=None,
|
||||||
duns_number=None,
|
duns_number=None,
|
||||||
):
|
):
|
||||||
# TODO: Fetch the contracts CSV and transform them into dictionaries.
|
response = self._get(
|
||||||
# https://docs.python.org/3/library/csv.html#csv.DictReader
|
"wawf_interface.returnContractList",
|
||||||
raise NotImplementedError()
|
pContract=contract_number,
|
||||||
|
pDelivery_Order=delivery_order,
|
||||||
|
pCage_Code=cage_code,
|
||||||
|
pDuns_Number=duns_number,
|
||||||
|
pUserName=self.user_name,
|
||||||
|
pUser_Role=self.user_role,
|
||||||
|
)
|
||||||
|
lines = response.text.replace("<br />", "").split("\n")
|
||||||
|
return list(DictReader(lines))
|
||||||
|
|
||||||
def get_contract(self, contract_number, status):
|
def get_contract(self, contract_number, status):
|
||||||
# TODO: Fetch the contract XML and transform it into a dictionary.
|
response = self._get(
|
||||||
# https://docs.python.org/3.7/library/xml.etree.elementtree.html
|
"pds_contract_interface.get_xml_doc",
|
||||||
raise NotImplementedError()
|
pContract=contract_number,
|
||||||
|
pStatus=status,
|
||||||
|
)
|
||||||
|
if response.text.startswith("No data found"):
|
||||||
|
return None
|
||||||
|
return ET.fromstring(response.text)
|
||||||
|
|
||||||
|
def get_clins(self, record_key, clins, cage_code="", duns_number=""):
|
||||||
|
response = self._get(
|
||||||
|
"wawf_interface.returnclinXML",
|
||||||
|
pCage_Code=cage_code,
|
||||||
|
pDuns_Number=duns_number,
|
||||||
|
pUserName=self.user_name,
|
||||||
|
pUser_Role=self.user_role,
|
||||||
|
pRecord_key=record_key,
|
||||||
|
pClins=clins,
|
||||||
|
)
|
||||||
|
# TODO: Parse XML, similar to `get_contract`
|
||||||
|
return response
|
||||||
|
Loading…
x
Reference in New Issue
Block a user