126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
from grocy.request import Request
|
|
from six.moves.urllib.parse import quote as encode_entity
|
|
import re
|
|
from copy import deepcopy
|
|
from grocy.conf import Configuration
|
|
import logging
|
|
|
|
|
|
class Entity(object):
|
|
RESOURCE_URL_TEMPLATE = '{domain}/api/objects/{entity}/{objectId}'
|
|
FIND_RESOURCE_BY_NAME = '{domain}/api/objects/{entity}/search/{name}'
|
|
COLLECTION_URL_TEMPLATE = '{domain}/api/objects/{entity}'
|
|
SCHEMA_URL_TEMPLATE = '{domain}/api/openapi/specification'
|
|
SCHEMA_MODEL_MAP = {
|
|
'products': 'Product',
|
|
'batteries': 'Battery',
|
|
'stock': 'StockEntry',
|
|
'product_groups': 'ProductGroup',
|
|
'locations': 'Location',
|
|
'quantity_units': 'QuantityUnit',
|
|
'recipes': 'Recipe'
|
|
}
|
|
|
|
def __init__(self, name, **props):
|
|
self.conf = Configuration()
|
|
self.conf.load()
|
|
self.name = name
|
|
self.__dict__.update(**props)
|
|
|
|
def get(self, id=None):
|
|
logger = logging.getLogger('entity.get')
|
|
if id:
|
|
url = self.RESOURCE_URL_TEMPLATE.format(domain=self.conf.domain, entity=self.name, objectId=id)
|
|
else:
|
|
url = self.COLLECTION_URL_TEMPLATE.format(domain=self.conf.domain, entity=self.name)
|
|
|
|
request = Request('get', url)
|
|
try:
|
|
return request.send()
|
|
except Exception as e:
|
|
logger.error(e)
|
|
raise e
|
|
|
|
def find(self, query):
|
|
logger = logging.getLogger('entity.find')
|
|
found_entities = []
|
|
try:
|
|
entities = self.get()
|
|
for entity in entities:
|
|
for prop, value in query.items():
|
|
regex = re.compile(r'{}'.format(value))
|
|
if regex.search(entity[prop]):
|
|
found_entities.append(entity)
|
|
|
|
if len(found_entities) == 0:
|
|
return []
|
|
|
|
return found_entities
|
|
except Exception as e:
|
|
logger.error(e)
|
|
raise e
|
|
|
|
def find_by_name(self, query):
|
|
logger = logging.getLogger('entity.find_by_name')
|
|
try:
|
|
url = self.FIND_RESOURCE_BY_NAME.format(domain=self.conf.domain, entity=self.name, name=encode_entity(query))
|
|
request = Request('get', url)
|
|
return request.send()
|
|
except Exception as e:
|
|
logger.error(e)
|
|
raise e
|
|
|
|
def create(self, entity):
|
|
logger = logging.getLogger('entity.add')
|
|
url = self.COLLECTION_URL_TEMPLATE.format(domain=self.conf.domain, entity=self.name)
|
|
|
|
for key, value in entity.items():
|
|
if type(value) == bool:
|
|
entity[key] = '1' if value else '0'
|
|
|
|
request = Request('post', url, resource=entity)
|
|
try:
|
|
return request.send()
|
|
except Exception as e:
|
|
logger.error(e)
|
|
raise e
|
|
|
|
def update(self, entity, id=None):
|
|
if id is None:
|
|
raise Exception('id property is required to update entity')
|
|
logger = logging.getLogger('entity.update')
|
|
url = self.RESOURCE_URL_TEMPLATE.format(domain=self.conf.domain, entity=self.name, objectId=id)
|
|
|
|
request = Request('put', url, resource=entity)
|
|
try:
|
|
return request.send()
|
|
except Exception as e:
|
|
logger.error(e)
|
|
raise e
|
|
|
|
def delete(self, id=None):
|
|
if id is None:
|
|
raise Exception('id property is required to delete entity')
|
|
logger = logging.getLogger('entity.delete')
|
|
url = self.RESOURCE_URL_TEMPLATE.format(domain=self.conf.domain, entity=self.name, objectId=id)
|
|
|
|
request = Request('delete', url)
|
|
try:
|
|
return request.send()
|
|
except Exception as e:
|
|
logger.error(e)
|
|
raise e
|
|
|
|
@property
|
|
def schema(self):
|
|
logger = logging.getLogger('entity.schema')
|
|
try:
|
|
url = self.SCHEMA_URL_TEMPLATE.format(domain=self.conf.domain)
|
|
request = Request('get', url)
|
|
response = request.send()
|
|
schema_name = self.SCHEMA_MODEL_MAP[self.name]
|
|
return response['components']['schemas'][schema_name]
|
|
except Exception as e:
|
|
logger.error(e)
|
|
raise e
|