I am extremely new to API’s. Although I have been able to call what I wish into Postman but I am having trouble on the perspective side.
I am trying to be able to query a table from Emaint. currently I am using system.net.httpclient to post data from a URL. Since emaint uses a JWT token, I am unable to get a good response so far. (I am guaranteed to be doing something wrong)
Emaint api:
Has anyone had luck with this? If so, do you mind sending parts of your code? (Without the token of course)
You need to authenticate with the /GetLoginToken end-point - it will give you back a token. You'll then hold onto this token and pass it along with your future requests.
We've done integration with a previous version of the X4 API, the v2 doesn't look that much different.
This is the smallest nugget of code I can share (it will need to be modified to work with v2 and the context/globals stuff will need to be removed).
class BaseX4Client(object):
"""Base clase for REST API calls to eMaint X4."""
server = "https://x46.emaint.com"
endpointQuery = '/wc.dll?x3~api~!&q='
clientArguments = {'redirect_policy': 'ALWAYS'}
validContexts = ['c', 'g']
context = 'c'
c_client = system.net.httpClient(**clientArguments)
g_client = mussonindustrial.net.GatewayHTTPClient(client_id='eMaint', **clientArguments)
@property
def client(self):
if self.context == 'c':
return self.c_client
if self.context == 'g':
return self.g_client
def setContext(self, context):
"""Set the context (gateway/client) of the HTTP client."""
if context in self.validContexts:
logger.debugf('Setting HTTP client context to "%s"' % context)
self.context = context
else:
raise ValueError("Context '%s' is not valid" % context)
return self
def getContext(self):
"""Get the context of the HTTP client."""
return self.context
def getWebMethodURL(self, webmethod, params=None):
"""Returns an eMaint webmethod URL with optional parameters."""
url = self.server + self.endpointQuery + webmethod
if params:
url += '&' + urllib.urlencode(params)
return url
@staticmethod
def formatDate(date):
"""Reformat a date into the format eMaint expects."""
return date.toInstant().truncatedTo(ChronoUnit.MILLIS)
class eMaintLoginToken(BaseX4Client):
"""A token used for authenticating eMaint Queries."""
def __init__(self, useragent, username, password):
"""Arguments:
useragent: Useragent to use for login.
username: Username to use for login.
password: Password to use for login.
"""
self.useragent = useragent
self.username = username
self.password = password
@property
def token(self):
"""Read the token from the globals if it exists.
If the token does not exist, get a new token."""
if 'token' in self._getGlobals():
return self._getGlobals()['token']
else:
return self._getGlobals().setdefault('token', self._newToken())
@token.setter
def token(self, value):
"""Set the token into globals."""
self._getGlobals()['token'] = value
def _getGlobalID(self):
"""Get the unique global ID for the authentication token."""
return hashlib.sha256((self.useragent+self.username+self.password).encode()).hexdigest()
def _getGlobals(self):
"""Return the globals stored for the authentication token."""
geMaint = mussonindustrial.globals.getGlobals(globalNamespace)
gAuth = geMaint.setdefault('auth', {})
return gAuth.setdefault(self._getGlobalID(), {})
def refresh(self):
"""Refresh the authentication token."""
self._getGlobals()['token'] = self._newToken()
return self
def _newToken(self):
"""Get a new token from the web API."""
logger.infof("Requesting a new authentication token from eMaint...")
url = self.getWebMethodURL('GetLoginToken')
headers = {
'XT-UserAgent': self.useragent,
'Username': self.username,
'Password': self.password,
'DataFormat': 'JSON'
}
try:
response = self.client.get(url, data={}, headers=headers).getJson()
if response['valid'] == 'false':
logger.errorf('Failed to login to eMaint with. Check username and password.')
raise Exception('Failure to login to eMaint. Check username and password.')
else:
logger.infof("Successfully recieved a new authentication token from eMaint.")
return response['token']
except Exception as e:
logger.errorf('Invalid response from eMaint. Check internet connection and eMaint availability.\n{}'.format(e.toString()))
raise Exception('Invalid response from eMaint. Check internet connection and eMaint availability.\n{}'.format(e.toString()))
def __repr__(self):
return '%s("%s", "%s", "%s")' % (self.__class__.__name__, self.useragent, self.username, self.password)