# -*- coding: utf-8 -*-
"""
mygeotab.api
~~~~~~~~~~~~
Public objects and methods wrapping the MyGeotab API.
"""
from __future__ import unicode_literals
import copy
import re
import ssl
import sys
from urllib.parse import urlparse
import requests
import urllib3
from requests.adapters import HTTPAdapter
from requests.exceptions import Timeout
from urllib3.util.ssl_ import create_urllib3_context
from . import __title__, __version__
from .exceptions import AuthenticationException, MyGeotabException, TimeoutException
from .serializers import json_deserialize, json_serialize
DEFAULT_TIMEOUT = 300
[docs]
class API(object):
"""A simple and Pythonic wrapper for the MyGeotab API."""
[docs]
def __init__(
self,
username,
password=None,
database=None,
session_id=None,
server="my.geotab.com",
timeout=DEFAULT_TIMEOUT,
proxies=None,
cert=None,
):
"""Initialize the MyGeotab API object with credentials.
:param username: The username used for MyGeotab servers. Usually an email address.
:type username: str
:param password: The password associated with the username. Optional if `session_id` is provided.
:type password: str
:param database: The database or company name. Optional as this usually gets resolved upon authentication.
:type database: str
:param session_id: A session ID, assigned by the server.
:type session_id: str
:param server: The server ie. my23.geotab.com. Optional as this usually gets resolved upon authentication.
:type server: str or None
:param timeout: The timeout to make the call, in seconds. By default, this is 300 seconds (or 5 minutes).
:type timeout: float or None
:param proxies: The proxies dictionary to apply to the request.
:type proxies: dict or None
:param cert: The path to client certificate. A single path to .pem file or a Tuple (.cer file, .key file).
:type cert: str or Tuple or None
:raise Exception: Raises an Exception if a username, or one of the session_id or password is not provided.
"""
if username is None:
raise Exception("`username` cannot be None")
if password is None and session_id is None:
raise Exception("`password` and `session_id` must not both be None")
self.credentials = Credentials(
username=username, session_id=session_id, database=database, server=server, password=password
)
self.timeout = timeout
self._proxies = proxies
self.__reauthorize_count = 0
self._cert = cert
@property
def _server(self):
if not self.credentials.server:
self.credentials.server = "my.geotab.com"
return self.credentials.server
@property
def _is_verify_ssl(self):
"""Whether or not SSL be verified.
:rtype: bool
:return: True if the calls are being made locally.
"""
return not any(s in get_api_url(self._server) for s in ["127.0.0.1", "localhost"])
[docs]
def call(self, method, **parameters):
"""Makes a call to the API.
:param method: The method name.
:type method: str
:param parameters: Additional parameters to send (for example, search=dict(id='b123') ).
:raise MyGeotabException: Raises when an exception occurs on the MyGeotab server.
:raise TimeoutException: Raises when the request does not respond after some time.
:return: The results from the server.
:rtype: dict or list
"""
if method is None:
raise Exception("A method name must be specified")
params = process_parameters(parameters)
if self.credentials and not self.credentials.session_id:
self.authenticate()
if "credentials" not in params and self.credentials.session_id:
params["credentials"] = self.credentials.get_param()
try:
result = _query(
self._server,
method,
params,
self.timeout,
verify_ssl=self._is_verify_ssl,
proxies=self._proxies,
cert=self._cert,
)
if result is not None:
self.__reauthorize_count = 0
return result
except MyGeotabException as exception:
if exception.name == "InvalidUserException" or (
exception.name == "DbUnavailableException" and "Initializing" in exception.message
):
if self.__reauthorize_count == 0 and self.credentials.password:
self.__reauthorize_count += 1
self.authenticate()
return self.call(method, **parameters)
else:
raise AuthenticationException(
self.credentials.username, self.credentials.database, self.credentials.server
) from exception
raise
[docs]
def multi_call(self, calls):
"""Performs a multi-call to the API.
:param calls: A list of call 2-tuples with method name and params
(for example, ('Get', dict(typeName='Trip')) ).
:type calls: list((str, dict))
:raise MyGeotabException: Raises when an exception occurs on the MyGeotab server.
:raise TimeoutException: Raises when the request does not respond after some time.
:return: The results from the server.
:rtype: list
"""
formatted_calls = [dict(method=call[0], params=call[1] if len(call) > 1 else {}) for call in calls]
return self.call("ExecuteMultiCall", calls=formatted_calls)
[docs]
def get(self, type_name, **parameters):
"""Gets entities using the API. Shortcut for using call() with the 'Get' method.
:param type_name: The type of entity.
:type type_name: str
:param parameters: Additional parameters to send. A parameter called `resultsLimit`
or `results_limit` will limit the number of entries returned. A
`search` parameter can further limit results, for example
search=dict(id='b123'). If a parameter called `search` is
omitted, any additional parameters are automatically added
to a `search` dictionary. This simplifies basic usage.
The following are equivalent calls:
api.get("Device", search={"id":"b2"})
api.get("Device", id="b2)
:raise MyGeotabException: Raises when an exception occurs on the MyGeotab server.
:raise TimeoutException: Raises when the request does not respond after some time.
:return: The results from the server.
:rtype: list
"""
if parameters:
# Detect resultsLimit if passed camelCase or python_case and
# remove from parameters (otherwise they will become part of search)
results_limit = parameters.get("resultsLimit")
if results_limit is not None:
del parameters["resultsLimit"]
else:
results_limit = parameters.get("results_limit")
if results_limit is not None:
del parameters["results_limit"]
if "search" in parameters:
parameters.update(parameters["search"])
del parameters["search"]
parameters = dict(search=parameters, resultsLimit=results_limit)
return self.call("Get", type_name=type_name, **parameters)
[docs]
def add(self, type_name, entity):
"""Adds an entity using the API. Shortcut for using call() with the 'Add' method.
:param type_name: The type of entity.
:type type_name: str
:param entity: The entity to add.
:type entity: dict
:raise MyGeotabException: Raises when an exception occurs on the MyGeotab server.
:raise TimeoutException: Raises when the request does not respond after some time.
:return: The id of the object added.
:rtype: str
"""
return self.call("Add", type_name=type_name, entity=entity)
[docs]
def set(self, type_name, entity):
"""Sets an entity using the API. Shortcut for using call() with the 'Set' method.
:param type_name: The type of entity.
:type type_name: str
:param entity: The entity to set.
:type entity: dict
:raise MyGeotabException: Raises when an exception occurs on the MyGeotab server.
:raise TimeoutException: Raises when the request does not respond after some time.
"""
return self.call("Set", type_name=type_name, entity=entity)
[docs]
def remove(self, type_name, entity):
"""Removes an entity using the API. Shortcut for using call() with the 'Remove' method.
:param type_name: The type of entity.
:type type_name: str
:param entity: The entity to remove.
:type entity: dict
:raise MyGeotabException: Raises when an exception occurs on the MyGeotab server.
:raise TimeoutException: Raises when the request does not respond after some time.
"""
return self.call("Remove", type_name=type_name, entity=entity)
[docs]
def authenticate(self):
"""Authenticates against the API server.
:raise AuthenticationException: Raises if there was an issue with authenticating or logging in.
:raise MyGeotabException: Raises when an exception occurs on the MyGeotab server.
:raise TimeoutException: Raises when the request does not respond after some time.
:return: A Credentials object with a session ID created by the server.
:rtype: Credentials
"""
try:
if self.credentials.session_id and not self.credentials.password:
# Extend the session if only the session ID is present
extend_session_data = dict(
database=self.credentials.database,
userName=self.credentials.username,
sessionId=self.credentials.session_id,
)
_query(
self._server,
"ExtendSession",
extend_session_data,
self.timeout,
verify_ssl=self._is_verify_ssl,
proxies=self._proxies,
cert=self._cert,
)
return self.credentials
auth_data = dict(
database=self.credentials.database,
userName=self.credentials.username,
password=self.credentials.password,
)
result = _query(
self._server,
"Authenticate",
auth_data,
self.timeout,
verify_ssl=self._is_verify_ssl,
proxies=self._proxies,
cert=self._cert,
)
if result:
if "path" not in result and self.credentials.session_id:
# Session was extended
return self.credentials
new_server = result["path"]
server = self.credentials.server
if new_server != "ThisServer":
server = new_server
credentials = result["credentials"]
self.credentials = Credentials(
credentials["userName"], credentials["sessionId"], credentials["database"], server
)
return self.credentials
except MyGeotabException as exception:
if exception.name == "InvalidUserException" or (
exception.name == "DbUnavailableException"
and ("Initializing" in exception.message or "UnknownDatabase" in exception.message)
):
raise AuthenticationException(
self.credentials.username, self.credentials.database, self.credentials.server
) from exception
raise
[docs]
@staticmethod
def from_credentials(credentials):
"""Returns a new API object from an existing Credentials object.
:param credentials: The existing saved credentials.
:type credentials: Credentials
:return: A new API object populated with MyGeotab credentials.
:rtype: API
"""
return API(
username=credentials.username,
password=credentials.password,
database=credentials.database,
session_id=credentials.session_id,
server=credentials.server,
)
[docs]
class Credentials(object):
"""The MyGeotab Credentials object."""
[docs]
def __init__(self, username, session_id, database, server, password=None):
"""Initialize the Credentials object.
:param username: The username used for MyGeotab servers. Usually an email address.
:type username: str
:param session_id: A session ID, assigned by the server.
:type session_id: str
:param database: The database or company name. Optional as this usually gets resolved upon authentication.
:type database: str or None
:param server: The server ie. my23.geotab.com. Optional as this usually gets resolved upon authentication.
:type server: str or None
:param password: The password associated with the username. Optional if `session_id` is provided.
:type password: str or None
"""
self.username = username
self.session_id = session_id
self.database = database
self.server = server
self.password = password
def __str__(self):
return "{0} @ {1}/{2}".format(self.username, self.server, self.database)
def __repr__(self):
return "Credentials(username={username}, database={database})".format(
username=self.username, database=self.database
)
[docs]
def get_param(self):
"""A simple representation of the credentials object for passing into the API.authenticate() server call.
:return: The simple credentials object for use by API.authenticate().
:rtype: dict
"""
return dict(userName=self.username, sessionId=self.session_id, database=self.database)
class GeotabHTTPAdapter(HTTPAdapter):
"""HTTP adapter to force use of TLS 1.2 for HTTPS connections."""
def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
ssl_context = create_urllib3_context(ssl_version=ssl.PROTOCOL_TLS)
ssl_context.options |= ssl.OP_NO_SSLv2
ssl_context.options |= ssl.OP_NO_SSLv3
ssl_context.options |= ssl.OP_NO_TLSv1
ssl_context.options |= ssl.OP_NO_TLSv1_1
self.poolmanager = urllib3.poolmanager.PoolManager(
num_pools=connections, maxsize=maxsize, block=block, ssl_context=ssl_context, **pool_kwargs
)
def _query(server, method, parameters, timeout=DEFAULT_TIMEOUT, verify_ssl=True, proxies=None, cert=None):
"""Formats and performs the query against the API.
:param server: The MyGeotab server.
:type server: str
:param method: The method name.
:type method: str
:param parameters: The parameters to send with the query.
:type parameters: dict
:param timeout: The timeout to make the call, in seconds. By default, this is 300 seconds (or 5 minutes).
:type timeout: float
:param verify_ssl: If True, verify the SSL certificate. It's recommended not to modify this.
:type verify_ssl: bool
:param proxies: The proxies dictionary to apply to the request.
:type proxies: dict or None
:param cert: The path to client certificate. A single path to .pem file or a Tuple (.cer file, .pem file)
:type cert: str or Tuple or None
:raise MyGeotabException: Raises when an exception occurs on the MyGeotab server.
:raise TimeoutException: Raises when the request does not respond after some time.
:raise urllib2.HTTPError: Raises when there is an HTTP status code that indicates failure.
:return: The JSON-decoded result from the server.
"""
api_endpoint = get_api_url(server)
params = dict(id=-1, method=method, params=parameters or {})
headers = get_headers()
with requests.Session() as session:
session.mount("https://", GeotabHTTPAdapter())
if cert:
session.cert = cert
try:
response = session.post(
api_endpoint,
data=json_serialize(params),
headers=headers,
allow_redirects=True,
timeout=timeout,
verify=verify_ssl,
proxies=proxies,
)
except Timeout as exc:
raise TimeoutException(server) from exc
response.raise_for_status()
content_type = response.headers.get("Content-Type")
if content_type and "application/json" not in content_type.lower():
return response.text
return _process(json_deserialize(response.text))
def _process(data):
"""Processes the returned JSON from the server.
:param data: The JSON data in dict form.
:raise MyGeotabException: Raises when a server exception was encountered.
:return: The result data.
"""
if data:
if "error" in data:
raise MyGeotabException(data["error"])
if "result" in data:
return data["result"]
return data
def server_call(method, server, timeout=DEFAULT_TIMEOUT, verify_ssl=True, proxies=None, **parameters):
"""Makes a call to an un-authenticated method on a server
:param method: The method name.
:type method: str
:param server: The MyGeotab server.
:type server: str
:param timeout: The timeout to make the call, in seconds. By default, this is 300 seconds (or 5 minutes).
:type timeout: float
:param verify_ssl: If True, verify the SSL certificate. It's recommended not to modify this.
:type verify_ssl: bool
:param proxies: The proxies dictionary to apply to the request.
:type proxies: dict or None
:param parameters: Additional parameters to send (for example, search=dict(id='b123') ).
:raise MyGeotabException: Raises when an exception occurs on the MyGeotab server.
:raise TimeoutException: Raises when the request does not respond after some time.
:return: The result from the server.
"""
if method is None:
raise Exception("A method name must be specified")
if server is None:
raise Exception("A server (eg. my3.geotab.com) must be specified")
parameters = process_parameters(parameters)
return _query(server, method, parameters, timeout=timeout, verify_ssl=verify_ssl, proxies=proxies)
def process_parameters(parameters):
"""Allows the use of Pythonic-style parameters with underscores instead of camel-case.
:param parameters: The parameters object.
:type parameters: dict
:return: The processed parameters.
:rtype: dict
"""
if not parameters:
return {}
params = copy.copy(parameters)
for param_name in parameters:
value = parameters[param_name]
server_param_name = re.sub(r"_(\w)", lambda m: m.group(1).upper(), param_name)
if isinstance(value, dict):
value = process_parameters(value)
params[server_param_name] = value
if server_param_name != param_name:
del params[param_name]
return params
def get_api_url(server):
"""Formats the server URL properly in order to query the API.
:return: A valid MyGeotab API request URL.
:rtype: str
"""
parsed = urlparse(server)
base_url = parsed.netloc if parsed.netloc else parsed.path
base_url.replace("/", "")
return "https://" + base_url + "/apiv1"
def get_headers():
"""Gets the request headers.
:return: The user agent
:rtype: dict
"""
return {
"Content-type": "application/json; charset=UTF-8",
"User-Agent": "Python/{py_version[0]}.{py_version[1]} {title}/{version}".format(
py_version=sys.version_info, title=__title__, version=__version__
),
}
__all__ = ["API", "Credentials", "MyGeotabException", "AuthenticationException"]