545 lines
17 KiB
Python
545 lines
17 KiB
Python
# -*- coding: utf-8; -*-
|
|
################################################################################
|
|
#
|
|
# pyCOREPOS -- Python Interface to CORE POS
|
|
# Copyright © 2018-2023 Lance Edgar
|
|
#
|
|
# This file is part of pyCOREPOS.
|
|
#
|
|
# pyCOREPOS is free software: you can redistribute it and/or modify it under
|
|
# the terms of the GNU General Public License as published by the Free
|
|
# Software Foundation, either version 3 of the License, or (at your option)
|
|
# any later version.
|
|
#
|
|
# pyCOREPOS is distributed in the hope that it will be useful, but WITHOUT ANY
|
|
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
|
|
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
|
|
# details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License along with
|
|
# pyCOREPOS. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
################################################################################
|
|
"""
|
|
CORE-POS webservices API
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
|
|
import requests
|
|
from requests.auth import HTTPDigestAuth
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class CoreAPIError(Exception):
|
|
"""
|
|
Base class for errors coming from the CORE API proper.
|
|
"""
|
|
|
|
def __init__(self, message):
|
|
self.message = message
|
|
|
|
def __str__(self):
|
|
return "CORE API returned an error: {}".format(self.message)
|
|
|
|
|
|
class CoreWebAPI(object):
|
|
"""
|
|
Client implementation for the CORE webservices API.
|
|
|
|
:param str url: URL to the CORE webservices API,
|
|
e.g. ``'http://localhost/fannie/ws/'``
|
|
|
|
:param bool verify: How to handle certificate validation for HTTPS
|
|
URLs. This value is passed as-is to the ``requests`` library,
|
|
so see those docs for more info. The default value for this is
|
|
``True`` because the assumption is that security should be on
|
|
by default. Set it to ``False`` in order to disable validation
|
|
entirely, e.g. for self-signed certs. (This may also be needed
|
|
for basic HTTP URLs?) Other values may be possible also; again
|
|
see the ``requests`` docs for more info.
|
|
|
|
:param htdigest_username: Username for htdigest authentication, if
|
|
applicable.
|
|
|
|
:param htdigest_password: Password for htdigest authentication, if
|
|
applicable.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
url,
|
|
verify=True,
|
|
htdigest_username=None,
|
|
htdigest_password=None,
|
|
):
|
|
self.url = url
|
|
self.verify = verify
|
|
|
|
self.session = requests.Session()
|
|
|
|
if htdigest_username and htdigest_password:
|
|
self.session.auth = HTTPDigestAuth(htdigest_username,
|
|
htdigest_password)
|
|
|
|
def post(self, params, method=None):
|
|
"""
|
|
Issue a POST request to the API, with the given ``params``. If not
|
|
specified, ``method`` will be CORE's ``FannieEntity`` webservice.
|
|
"""
|
|
if not method:
|
|
method = 'FannieEntity'
|
|
if '\\' not in method:
|
|
method = r'\COREPOS\Fannie\API\webservices\{}'.format(method)
|
|
|
|
payload = {
|
|
'jsonrpc': '2.0',
|
|
'method': method,
|
|
'params': params,
|
|
# we're not dealing with async here, so KISS for this 'id'
|
|
# https://stackoverflow.com/questions/4390369/json-rpc-how-can-one-make-a-unique-id#comment4786119_4391070
|
|
'id': 1,
|
|
}
|
|
|
|
response = self.session.post(self.url, data=json.dumps(payload),
|
|
verify=self.verify)
|
|
response.raise_for_status()
|
|
return response
|
|
|
|
def parse_response(self, response, method=None):
|
|
"""
|
|
Generic method to "parse" a response from the API. Really this just
|
|
converts the JSON to a dict (etc.), and then checks for error. If an
|
|
error is found in the response, it will be raised here.
|
|
"""
|
|
try:
|
|
js = response.json()
|
|
except:
|
|
raise CoreAPIError("Received invalid response: {}".format(response.content))
|
|
|
|
if 'error' in js:
|
|
raise CoreAPIError(js['error'])
|
|
|
|
# note, the result data format may depend on the API method involved
|
|
if method == 'FannieMember':
|
|
return js['result']
|
|
|
|
# assuming typical FannieEntity result here
|
|
assert set(js.keys()) == set(['jsonrpc', 'id', 'result'])
|
|
assert set(js['result'].keys()) == set(['result'])
|
|
return js['result']['result']
|
|
|
|
def get_member_types(self):
|
|
"""
|
|
Fetch all Member Type records from CORE.
|
|
|
|
:returns: A (potentially empty) list of member type dict records.
|
|
"""
|
|
params = {
|
|
'entity': 'Memtype',
|
|
'submethod': 'get',
|
|
'columns': {},
|
|
}
|
|
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return [json.loads(rec) for rec in result]
|
|
|
|
def get_members(self):
|
|
"""
|
|
Fetch all Member records from CORE.
|
|
|
|
:returns: A (potentially empty) list of member dict records.
|
|
"""
|
|
params = {
|
|
'method': 'get',
|
|
'cardNo': None,
|
|
}
|
|
response = self.post(params, method='FannieMember')
|
|
result = self.parse_response(response, method='FannieMember')
|
|
return result
|
|
|
|
def get_member(self, cardNo):
|
|
"""
|
|
Fetch an existing Member record from CORE.
|
|
|
|
:returns: Either a member dict record, or ``None``.
|
|
"""
|
|
params = {
|
|
'cardNo': cardNo,
|
|
'method': 'get',
|
|
}
|
|
response = self.post(params, method='FannieMember')
|
|
result = self.parse_response(response, method='FannieMember')
|
|
if result:
|
|
return result
|
|
|
|
def set_member(self, cardNo, **kwargs):
|
|
"""
|
|
Update an existing Member record in CORE.
|
|
|
|
:returns: Boolean indicating success of the operation.
|
|
|
|
.. warning::
|
|
Only simple updates have been attempted thus far; have yet to try
|
|
creation or deletion. Neither of those should be expected to work.
|
|
"""
|
|
kwargs['cardNo'] = cardNo
|
|
params = {
|
|
'cardNo': cardNo,
|
|
'method': 'set',
|
|
'member': kwargs,
|
|
}
|
|
response = self.post(params, method='FannieMember')
|
|
result = self.parse_response(response, method='FannieMember')
|
|
if result:
|
|
return result
|
|
|
|
def get_stores(self, **columns):
|
|
"""
|
|
Fetch some or all of Store records from CORE.
|
|
|
|
:returns: A (potentially empty) list of store dict records.
|
|
|
|
To fetch all stores::
|
|
|
|
api.get_stores()
|
|
|
|
To fetch only stores named "Headquarters"::
|
|
|
|
api.get_stores(description='Headquarters')
|
|
"""
|
|
params = {
|
|
'entity': 'Stores',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return [json.loads(rec) for rec in result]
|
|
|
|
def get_departments(self, **columns):
|
|
"""
|
|
Fetch some or all of Department records from CORE.
|
|
|
|
:returns: A (potentially empty) list of department dict records.
|
|
|
|
To fetch all departments::
|
|
|
|
api.get_departments()
|
|
|
|
To fetch only departments named "Grocery"::
|
|
|
|
api.get_departments(dept_name='Grocery')
|
|
"""
|
|
params = {
|
|
'entity': 'Departments',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return [json.loads(rec) for rec in result]
|
|
|
|
def get_department(self, dept_no, **columns):
|
|
"""
|
|
Fetch an existing Department record from CORE.
|
|
|
|
:returns: Either a department dict record, or ``None``.
|
|
"""
|
|
columns['dept_no'] = dept_no
|
|
params = {
|
|
'entity': 'Departments',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
if result:
|
|
if len(result) > 1:
|
|
log.warning("CORE API returned %s department results", len(result))
|
|
return json.loads(result[0])
|
|
|
|
def set_department(self, dept_no, **columns):
|
|
"""
|
|
Update an existing Department record in CORE.
|
|
|
|
:returns: Boolean indicating success of the operation.
|
|
|
|
.. note::
|
|
Currently this is being used to create a *new* department also. CORE's
|
|
``departments`` table does not use auto-increment for its PK, which
|
|
means we must provide one even when creating; therefore this method
|
|
may be used for that.
|
|
"""
|
|
columns['dept_no'] = dept_no
|
|
params = {
|
|
'entity': 'Departments',
|
|
'submethod': 'set',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return json.loads(result)
|
|
|
|
def get_subdepartments(self, **columns):
|
|
"""
|
|
Fetch some or all of Subdepartment records from CORE.
|
|
|
|
:returns: A (potentially empty) list of subdepartment dict records.
|
|
|
|
To fetch all subdepartments::
|
|
|
|
api.get_subdepartments()
|
|
|
|
To fetch only subdepartments named "Grocery"::
|
|
|
|
api.get_subdepartments(subdept_name='Grocery')
|
|
"""
|
|
params = {
|
|
'entity': 'SubDepts',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return [json.loads(rec) for rec in result]
|
|
|
|
def get_subdepartment(self, subdept_no, **columns):
|
|
"""
|
|
Fetch an existing Subdepartment record from CORE.
|
|
|
|
:returns: Either a subdepartment dict record, or ``None``.
|
|
"""
|
|
columns['subdept_no'] = subdept_no
|
|
params = {
|
|
'entity': 'SubDepts',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
if result:
|
|
if len(result) > 1:
|
|
log.warning("CORE API returned %s subdepartment results", len(result))
|
|
return json.loads(result[0])
|
|
|
|
def set_subdepartment(self, subdept_no, **columns):
|
|
"""
|
|
Update an existing Subdepartment record in CORE.
|
|
|
|
:returns: Boolean indicating success of the operation.
|
|
|
|
.. note::
|
|
Currently this is being used to create a *new* subdepartment also. CORE's
|
|
``subdepartments`` table does not use auto-increment for its PK, which
|
|
means we must provide one even when creating; therefore this method
|
|
may be used for that.
|
|
"""
|
|
columns['subdept_no'] = subdept_no
|
|
params = {
|
|
'entity': 'SubDepts',
|
|
'submethod': 'set',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return json.loads(result)
|
|
|
|
def get_vendors(self, **columns):
|
|
"""
|
|
Fetch some or all of Vendor records from CORE.
|
|
|
|
:returns: A (potentially empty) list of vendor dict records.
|
|
|
|
To fetch all vendors::
|
|
|
|
api.get_vendors()
|
|
|
|
To fetch only vendors named "UNFI"::
|
|
|
|
api.get_vendors(vendorName='UNFI')
|
|
"""
|
|
params = {
|
|
'entity': 'Vendors',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return [json.loads(rec) for rec in result]
|
|
|
|
def get_vendor(self, vendorID, **columns):
|
|
"""
|
|
Fetch an existing Vendor record from CORE.
|
|
|
|
:returns: Either a vendor dict record, or ``None``.
|
|
"""
|
|
columns['vendorID'] = vendorID
|
|
params = {
|
|
'entity': 'Vendors',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
if result:
|
|
if len(result) > 1:
|
|
log.warning("CORE API returned %s vendor results", len(result))
|
|
return json.loads(result[0])
|
|
|
|
def set_vendor(self, vendorID, **columns):
|
|
"""
|
|
Update an existing Vendor record in CORE.
|
|
|
|
:returns: Boolean indicating success of the operation.
|
|
|
|
.. note::
|
|
Currently this is being used to create a *new* vendor also. CORE's
|
|
``vendors`` table does not use auto-increment for its PK, which
|
|
means we must provide one even when creating; therefore this method
|
|
may be used for that.
|
|
"""
|
|
columns['vendorID'] = vendorID
|
|
params = {
|
|
'entity': 'Vendors',
|
|
'submethod': 'set',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return json.loads(result)
|
|
|
|
def get_products(self, **columns):
|
|
"""
|
|
Fetch some or all of Product records from CORE.
|
|
|
|
:returns: A (potentially empty) list of product dict records.
|
|
|
|
To fetch all products::
|
|
|
|
api.get_products()
|
|
|
|
To fetch only products with brand name "Braggs"::
|
|
|
|
api.get_products(brand='Braggs')
|
|
"""
|
|
params = {
|
|
'entity': 'Products',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return [json.loads(rec) for rec in result]
|
|
|
|
def get_product(self, upc, **columns):
|
|
"""
|
|
Fetch an existing Product record from CORE.
|
|
|
|
:returns: Either a product dict record, or ``None``.
|
|
"""
|
|
columns['upc'] = upc
|
|
params = {
|
|
'entity': 'Products',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
if result:
|
|
if len(result) > 1:
|
|
log.warning("CORE API returned %s product results", len(result))
|
|
return json.loads(result[0])
|
|
|
|
def set_product(self, upc, **columns):
|
|
"""
|
|
Update an existing Product record in CORE.
|
|
|
|
:returns: Boolean indicating success of the operation.
|
|
|
|
.. note::
|
|
Currently this is being used to create a *new* product also. CORE's
|
|
``products`` table does not use auto-increment for its PK, which
|
|
means we must provide one even when creating; therefore this method
|
|
may be used for that.
|
|
"""
|
|
columns['upc'] = upc
|
|
params = {
|
|
'entity': 'Products',
|
|
'submethod': 'set',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return json.loads(result)
|
|
|
|
def get_vendor_items(self, **columns):
|
|
"""
|
|
Fetch some or all of VendorItem records from CORE.
|
|
|
|
:returns: A (potentially empty) list of vendor item dict records.
|
|
|
|
To fetch all vendor items::
|
|
|
|
api.get_vendor_items()
|
|
|
|
To fetch only products with brand name "Braggs"::
|
|
|
|
api.get_vendor_items(brand='Braggs')
|
|
"""
|
|
params = {
|
|
'entity': 'VendorItems',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return [json.loads(rec) for rec in result]
|
|
|
|
def get_vendor_item(self, sku, vendorID, **columns):
|
|
"""
|
|
Fetch an existing VendorItem record from CORE.
|
|
|
|
:returns: Either a vendor item dict record, or ``None``.
|
|
"""
|
|
columns['sku'] = sku
|
|
columns['vendorID'] = vendorID
|
|
params = {
|
|
'entity': 'VendorItems',
|
|
'submethod': 'get',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
if result:
|
|
if len(result) > 1:
|
|
log.warning("CORE API returned %s VendorItem results", len(result))
|
|
return json.loads(result[0])
|
|
|
|
def set_vendor_item(self, sku, vendorID, **columns):
|
|
"""
|
|
Update an existing VendorItem record in CORE.
|
|
|
|
:returns: Boolean indicating success of the operation.
|
|
|
|
.. note::
|
|
Currently this is being used to create a *new* product also. CORE's
|
|
``vendorItems`` table does not use auto-increment for its PK, which
|
|
means we must provide one even when creating; therefore this method
|
|
may be used for that.
|
|
"""
|
|
columns['sku'] = sku
|
|
columns['vendorID'] = vendorID
|
|
params = {
|
|
'entity': 'VendorItems',
|
|
'submethod': 'set',
|
|
'columns': columns,
|
|
}
|
|
response = self.post(params)
|
|
result = self.parse_response(response)
|
|
return json.loads(result)
|