Source code for pysnow.client

# -*- coding: utf-8 -*-

import requests
import warnings
from pysnow import request
from pysnow.exceptions import InvalidUsage

warnings.simplefilter("always", DeprecationWarning)


[docs]class Client(object): def __init__(self, instance=None, host=None, user=None, password=None, raise_on_empty=True, request_params=None, use_ssl=True, session=None): """Creates a client ready to handle requests :param instance: instance name, used to construct host :param host: host can be passed as an alternative to instance :param user: username :param password: password :param raise_on_empty: whether or not to raise an exception on 404 (no matching records) :param request_params: request params to send with requests :param use_ssl: Enable or disable SSL :param session: a requests session object """ if (host and instance) is not None: raise InvalidUsage("Instance and host are mutually exclusive, you cannot use both.") if type(use_ssl) is not bool: raise InvalidUsage("Argument use_ssl must be of type bool") if type(raise_on_empty) is not bool: raise InvalidUsage("Argument raise_on_empty must be of type bool") if not (host or instance): raise InvalidUsage("You must supply an instance name or a host") if not (user and password) and not session: raise InvalidUsage("You must provide either username and password or a session object") elif (user and session) is not None: raise InvalidUsage("Provide either username and password or a session, not both.") if request_params is not None: if isinstance(request_params, dict): self.request_params = request_params else: raise InvalidUsage("Request params must be of type dict") else: self.request_params = {} self.instance = instance self.host = host self._user = user self._password = password self.raise_on_empty = raise_on_empty self.use_ssl = use_ssl self.base_url = self._get_base_url() self.session = self._get_session(session) def _get_session(self, session): if not session: s = requests.Session() s.auth = requests.auth.HTTPBasicAuth(self._user, self._password) else: s = session s.headers.update({'content-type': 'application/json', 'accept': 'application/json'}) return s def _get_base_url(self): if self.instance is not None: self.host = "%s.service-now.com" % self.instance if self.use_ssl is True: return "https://%s" % self.host return "http://%s" % self.host def _request(self, method, table, **kwargs): """Creates and returns a new `Request` object, takes some basic settings from the `Client` object and passes along to the `Request` constructor :param method: HTTP method :param table: Table to operate on :param kwargs: Keyword arguments passed along to `Request` :return: `Request` object """ return request.Request(method, table, request_params=self.request_params, raise_on_empty=self.raise_on_empty, session=self.session, instance=self.instance, base_url=self.base_url, **kwargs)
[docs] def query(self, table, **kwargs): """Query (GET) request wrapper. :param table: table to perform query on :param kwargs: Keyword arguments passed along to `Request` :return: `Request` object """ return self._request('GET', table, **kwargs)
[docs] def insert(self, table, payload, **kwargs): """Insert (POST) request wrapper :param table: table to insert on :param payload: update payload (dict) :param kwargs: Keyword arguments passed along to `Request` :return: New record content """ r = self._request('POST', table, **kwargs) return r.insert(payload)