Module pygrocy.grocy_api_client
Expand source code
import base64
import json
import logging
from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Optional
from urllib.parse import urljoin
import requests
from pydantic import BaseModel, Extra, Field, root_validator, validator
from pydantic.schema import date
from pygrocy import EntityType
from pygrocy.utils import grocy_datetime_str, localize_datetime, parse_date
from .errors import GrocyError
DEFAULT_PORT_NUMBER = 9192
_LOGGER = logging.getLogger(__name__)
_LOGGER.setLevel(logging.INFO)
def _field_not_empty_validator(field_name: str):
"""Reusable Pydantic field pre-validator to convert empty str to None."""
return validator(field_name, allow_reuse=True, pre=True)(_none_if_empty_str)
def _none_if_empty_str(value: Any):
if isinstance(value, str) and value == "":
return None
return value
class ShoppingListItem(BaseModel):
id: int
product_id: Optional[int] = None
note: Optional[str] = None
amount: float
row_created_timestamp: datetime
shopping_list_id: int
done: int
class MealPlanResponse(BaseModel):
id: int
day: date
type: str
recipe_id: Optional[int] = None
recipe_servings: Optional[int] = None
note: Optional[str] = None
product_id: Optional[int] = None
product_amount: Optional[float] = None
product_qu_id: Optional[str] = None
row_created_timestamp: datetime
userfields: Optional[Dict] = None
section_id: Optional[int] = None
class RecipeDetailsResponse(BaseModel):
id: Optional[int] = None
name: str
description: Optional[str] = None
base_servings: int
desired_servings: int
picture_file_name: Optional[str]
row_created_timestamp: datetime
userfields: Optional[Dict] = None
class QuantityUnitData(BaseModel):
id: int
name: str
name_plural: str
description: Optional[str] = None
row_created_timestamp: datetime
class LocationData(BaseModel):
id: int
name: str
description: Optional[str] = None
row_created_timestamp: datetime
class ProductData(BaseModel):
id: int
name: str
description: Optional[str] = None
location_id: Optional[int] = None
product_group_id: Optional[int] = None
qu_id_stock: int
qu_id_purchase: int
picture_file_name: Optional[str] = None
allow_partial_units_in_stock: Optional[bool] = False
row_created_timestamp: datetime
min_stock_amount: Optional[float]
default_best_before_days: int
location_id_validator = _field_not_empty_validator("location_id")
product_group_id_validator = _field_not_empty_validator("product_group_id")
class ChoreData(BaseModel):
id: int
name: str
description: Optional[str] = None
period_type: str
period_config: Optional[str] = None
period_days: Optional[int] = 0
track_date_only: bool
rollover: bool
assignment_type: Optional[str] = None
assignment_config: Optional[str] = None
next_execution_assigned_to_user_id: Optional[int] = None
userfields: Optional[Dict]
next_execution_assigned_to_user_id_validator = _field_not_empty_validator(
"next_execution_assigned_to_user_id"
)
class UserDto(BaseModel):
id: int
username: str
first_name: Optional[str] = None
last_name: Optional[str] = None
display_name: Optional[str] = None
class CurrentChoreResponse(BaseModel):
chore_id: int
last_tracked_time: Optional[datetime] = None
next_estimated_execution_time: Optional[datetime] = None
class CurrentStockResponse(BaseModel):
product_id: int
amount: float
best_before_date: date
amount_opened: float
amount_aggregated: float
amount_opened_aggregated: float
is_aggregated_amount: bool
product: ProductData
class MissingProductResponse(BaseModel):
id: int
name: str
amount_missing: float
is_partly_in_stock: bool
class CurrentVolatilStockResponse(BaseModel):
due_products: Optional[List[CurrentStockResponse]] = None
overdue_products: Optional[List[CurrentStockResponse]] = None
expired_products: Optional[List[CurrentStockResponse]] = None
missing_products: Optional[List[MissingProductResponse]] = None
class ProductBarcodeData(BaseModel):
barcode: str
class ProductDetailsResponse(BaseModel):
last_purchased: Optional[date] = None
last_used: Optional[date] = None
stock_amount: float
stock_amount_opened: float
next_best_before_date: Optional[date] = None
last_price: Optional[float] = None
product: ProductData
quantity_unit_stock: QuantityUnitData
default_quantity_unit_purchase: QuantityUnitData
barcodes: Optional[List[ProductBarcodeData]] = Field(alias="product_barcodes")
location: Optional[LocationData] = None
class ChoreDetailsResponse(BaseModel):
chore: ChoreData
last_tracked: Optional[datetime] = None
next_estimated_execution_time: Optional[datetime] = None
track_count: int = 0
next_execution_assigned_user: Optional[UserDto] = None
last_done_by: Optional[UserDto] = None
class TransactionType(Enum):
PURCHASE = "purchase"
CONSUME = "consume"
INVENTORY_CORRECTION = "inventory-correction"
PRODUCT_OPENED = "product-opened"
class TaskCategoryDto(BaseModel):
id: int
name: str
description: Optional[str] = None
row_created_timestamp: datetime
class TaskResponse(BaseModel):
id: int
name: str
description: Optional[str] = None
due_date: Optional[date] = None
done: int
done_timestamp: Optional[datetime] = None
category_id: Optional[int] = None
category: Optional[TaskCategoryDto] = None
assigned_to_user_id: Optional[int] = None
assigned_to_user: Optional[UserDto] = None
userfields: Optional[Dict] = None
due_date_validator = _field_not_empty_validator("due_date")
category_id_validator = _field_not_empty_validator("category_id")
assigned_to_user_id_validator = _field_not_empty_validator("assigned_to_user_id")
class CurrentBatteryResponse(BaseModel):
id: int
last_tracked_time: Optional[datetime] = None
next_estimated_charge_time: Optional[datetime] = None
class BatteryData(BaseModel):
id: int
name: str
description: str
used_in: str
charge_interval_days: int
created_timestamp: datetime = Field(alias="row_created_timestamp")
userfields: Optional[Dict] = None
class BatteryDetailsResponse(BaseModel):
battery: BatteryData
charge_cycles_count: int
last_charged: Optional[datetime] = None
last_tracked_time: Optional[datetime] = None
next_estimated_charge_time: Optional[datetime] = None
class MealPlanSectionResponse(BaseModel):
id: Optional[int] = None
name: str
sort_number: Optional[int] = None
row_created_timestamp: datetime
sort_number_validator = _field_not_empty_validator("sort_number")
class StockLogResponse(BaseModel):
id: int
product_id: int
amount: float
best_before_date: date
purchased_date: date
used_date: Optional[date] = None
spoiled: bool = False
stock_id: str
transaction_id: str
transaction_type: TransactionType
class GrocyVersionDto(BaseModel):
version: str = Field(alias="Version")
release_date: date = Field(alias="ReleaseDate")
class SystemInfoDto(BaseModel):
grocy_version_info: GrocyVersionDto = Field(alias="grocy_version")
php_version: str
sqlite_version: str
os: str
client: str
class SystemTimeDto(BaseModel):
timezone: str
time_local: datetime
time_local_sqlite3: datetime
time_utc: datetime
timestamp: int
class SystemConfigDto(BaseModel, extra=Extra.allow):
username: str = Field(alias="USER_USERNAME")
base_path: str = Field(alias="BASE_PATH")
base_url: str = Field(alias="BASE_URL")
mode: str = Field(alias="MODE")
default_locale: str = Field(alias="DEFAULT_LOCALE")
locale: str = Field(alias="LOCALE")
currency: str = Field(alias="CURRENCY")
feature_flags: Dict[str, Any]
@root_validator(pre=True)
def feature_flags_root_validator(cls, fields: Dict[str, Any]) -> Dict[str, Any]:
"""Pydantic root validator to add all "FEATURE_FLAG_" settings to Dict."""
features: Dict[str, Any] = {}
class_defined_fields = cls.__fields__.values()
for field, value in fields.items():
if field.startswith("FEATURE_FLAG_") and field not in class_defined_fields:
features[field] = value
fields["feature_flags"] = features
return fields
def _enable_debug_mode():
_LOGGER.setLevel(logging.DEBUG)
class GrocyApiClient(object):
def __init__(
self,
base_url,
api_key,
port: int = DEFAULT_PORT_NUMBER,
path: str = None,
verify_ssl=True,
debug=False,
):
if debug:
_enable_debug_mode()
if path:
self._base_url = "{}:{}/{}/api/".format(base_url, port, path)
else:
self._base_url = "{}:{}/api/".format(base_url, port)
_LOGGER.debug(f"generated base url: {self._base_url}")
self._api_key = api_key
self._verify_ssl = verify_ssl
if self._api_key == "demo_mode":
self._headers = {"accept": "application/json"}
else:
self._headers = {"accept": "application/json", "GROCY-API-KEY": api_key}
def _do_get_request(self, end_url: str, query_filters: List[str] = None):
req_url = urljoin(self._base_url, end_url)
params = None
if query_filters:
params = {"query[]": query_filters}
resp = requests.get(
req_url, verify=self._verify_ssl, headers=self._headers, params=params
)
_LOGGER.debug("-->\tGET /%s", end_url)
_LOGGER.debug("<--\t%d for /%s", resp.status_code, end_url)
_LOGGER.debug("\t\t%s", resp.content)
if resp.status_code >= 400:
raise GrocyError(resp)
if len(resp.content) > 0:
return resp.json()
def _do_post_request(self, end_url: str, data: dict):
req_url = urljoin(self._base_url, end_url)
resp = requests.post(
req_url, verify=self._verify_ssl, headers=self._headers, json=data
)
_LOGGER.debug("-->\tPOST /%s", end_url)
_LOGGER.debug("\t\t%s", data)
_LOGGER.debug("<--\t%d for /%s", resp.status_code, end_url)
_LOGGER.debug("\t\t%s", resp.content)
if resp.status_code >= 400:
raise GrocyError(resp)
if len(resp.content) > 0:
return resp.json()
def _do_put_request(self, end_url: str, data):
req_url = urljoin(self._base_url, end_url)
up_header = self._headers.copy()
up_header["accept"] = "*/*"
if isinstance(data, dict):
up_header["Content-Type"] = "application/json"
data = json.dumps(data)
else:
up_header["Content-Type"] = "application/octet-stream"
resp = requests.put(
req_url, verify=self._verify_ssl, headers=up_header, data=data
)
_LOGGER.debug("-->\tPUT /%s", end_url)
_LOGGER.debug("\t\t%s", data)
_LOGGER.debug("<--\t%d for /%s", resp.status_code, end_url)
_LOGGER.debug("\t\t%s", resp.content)
if resp.status_code >= 400:
raise GrocyError(resp)
if len(resp.content) > 0:
return resp.json()
def _do_delete_request(self, end_url: str):
req_url = urljoin(self._base_url, end_url)
resp = requests.delete(req_url, verify=self._verify_ssl, headers=self._headers)
_LOGGER.debug("-->\tDELETE /%s", end_url)
_LOGGER.debug("<--\t%d for /%s", resp.status_code, end_url)
_LOGGER.debug("\t\t%s", resp.content)
if resp.status_code >= 400:
raise GrocyError(resp)
if len(resp.content) > 0:
return resp.json()
def get_stock(self) -> List[CurrentStockResponse]:
parsed_json = self._do_get_request("stock")
if parsed_json:
return [CurrentStockResponse(**response) for response in parsed_json]
return []
def get_volatile_stock(self) -> CurrentVolatilStockResponse:
parsed_json = self._do_get_request("stock/volatile")
return CurrentVolatilStockResponse(**parsed_json)
def get_product(self, product_id) -> ProductDetailsResponse:
url = f"stock/products/{product_id}"
parsed_json = self._do_get_request(url)
if parsed_json:
return ProductDetailsResponse(**parsed_json)
def get_product_by_barcode(self, barcode) -> ProductDetailsResponse:
url = f"stock/products/by-barcode/{barcode}"
parsed_json = self._do_get_request(url)
if parsed_json:
return ProductDetailsResponse(**parsed_json)
def get_chores(self, query_filters: List[str] = None) -> List[CurrentChoreResponse]:
parsed_json = self._do_get_request("chores", query_filters)
if parsed_json:
return [CurrentChoreResponse(**chore) for chore in parsed_json]
return []
def get_chore(self, chore_id: int) -> ChoreDetailsResponse:
url = f"chores/{chore_id}"
parsed_json = self._do_get_request(url)
if parsed_json:
return ChoreDetailsResponse(**parsed_json)
def execute_chore(
self,
chore_id: int,
done_by: int = None,
tracked_time: datetime = None,
skipped: bool = False,
):
if tracked_time is None:
tracked_time = datetime.now()
localized_tracked_time = localize_datetime(tracked_time)
data = {
"tracked_time": grocy_datetime_str(localized_tracked_time),
"skipped": skipped,
}
if done_by is not None:
data["done_by"] = done_by
return self._do_post_request(f"chores/{chore_id}/execute", data)
def add_product(
self,
product_id,
amount: float,
price: float,
best_before_date: datetime = None,
transaction_type: TransactionType = TransactionType.PURCHASE,
):
data = {
"amount": amount,
"transaction_type": transaction_type.value,
"price": price,
}
if best_before_date is not None:
data["best_before_date"] = best_before_date.strftime("%Y-%m-%d")
return self._do_post_request(f"stock/products/{product_id}/add", data)
def consume_product(
self,
product_id: int,
amount: float = 1,
spoiled: bool = False,
transaction_type: TransactionType = TransactionType.CONSUME,
allow_subproduct_substitution: bool = False,
):
data = {
"amount": amount,
"spoiled": spoiled,
"transaction_type": transaction_type.value,
"allow_subproduct_substitution": allow_subproduct_substitution,
}
self._do_post_request(f"stock/products/{product_id}/consume", data)
def open_product(
self,
product_id: int,
amount: float = 1,
allow_subproduct_substitution: bool = False,
):
data = {
"amount": amount,
"allow_subproduct_substitution": allow_subproduct_substitution,
}
self._do_post_request(f"stock/products/{product_id}/open", data)
def consume_recipe(
self,
recipe_id: int,
):
self._do_post_request(f"recipes/{recipe_id}/consume", None)
def inventory_product(
self,
product_id: int,
new_amount: float,
best_before_date: datetime = None,
shopping_location_id: int = None,
location_id: int = None,
price: float = None,
):
data = {
"new_amount": new_amount,
}
if best_before_date is not None:
data["best_before_date"] = localize_datetime(best_before_date).strftime(
"%Y-%m-%d"
)
if shopping_location_id is not None:
data["shopping_location_id"] = shopping_location_id
if location_id is not None:
data["location_id"] = location_id
if price is not None:
data["price"] = price
parsed_json = self._do_post_request(
f"stock/products/{product_id}/inventory", data
)
if parsed_json:
stockLog = [StockLogResponse(**response) for response in parsed_json]
return stockLog[0]
def add_product_by_barcode(
self,
barcode: str,
amount: float,
price: float,
best_before_date: datetime = None,
) -> StockLogResponse:
data = {
"amount": amount,
"transaction_type": TransactionType.PURCHASE.value,
"price": price,
}
if best_before_date is not None:
data["best_before_date"] = localize_datetime(best_before_date).strftime(
"%Y-%m-%d"
)
parsed_json = self._do_post_request(
f"stock/products/by-barcode/{barcode}/add", data
)
if parsed_json:
stockLog = [StockLogResponse(**response) for response in parsed_json]
return stockLog[0]
def consume_product_by_barcode(
self, barcode: str, amount: float = 1, spoiled: bool = False
):
data = {
"amount": amount,
"spoiled": spoiled,
"transaction_type": TransactionType.CONSUME.value,
}
parsed_json = self._do_post_request(
f"stock/products/by-barcode/{barcode}/consume", data
)
if parsed_json:
stockLog = [StockLogResponse(**response) for response in parsed_json]
return stockLog[0]
def inventory_product_by_barcode(
self,
barcode: str,
new_amount: float,
best_before_date: datetime = None,
location_id: int = None,
price: float = None,
):
data = {
"new_amount": new_amount,
}
if best_before_date is not None:
data["best_before_date"] = localize_datetime(best_before_date).strftime(
"%Y-%m-%d"
)
if location_id is not None:
data["location_id"] = location_id
if price is not None:
data["price"] = price
parsed_json = self._do_post_request(
f"stock/products/by-barcode/{barcode}/inventory", data
)
if parsed_json:
stockLog = [StockLogResponse(**response) for response in parsed_json]
return stockLog[0]
def get_shopping_list(
self, query_filters: List[str] = None
) -> List[ShoppingListItem]:
parsed_json = self._do_get_request("objects/shopping_list", query_filters)
if parsed_json:
return [ShoppingListItem(**response) for response in parsed_json]
return []
def add_missing_product_to_shopping_list(self, shopping_list_id: int = None):
data = None
if shopping_list_id:
data = {"list_id": shopping_list_id}
self._do_post_request("stock/shoppinglist/add-missing-products", data)
def add_product_to_shopping_list(
self,
product_id: int,
shopping_list_id: int = 1,
amount: float = 1,
quantity_unit_id: int = None,
):
data = {
"product_id": product_id,
"list_id": shopping_list_id,
"product_amount": amount,
}
if quantity_unit_id:
data["qu_id"] = quantity_unit_id
self._do_post_request("stock/shoppinglist/add-product", data)
def clear_shopping_list(self, shopping_list_id: int = 1):
data = {"list_id": shopping_list_id}
self._do_post_request("stock/shoppinglist/clear", data)
def remove_product_in_shopping_list(
self, product_id: int, shopping_list_id: int = 1, amount: float = 1
):
data = {
"product_id": product_id,
"list_id": shopping_list_id,
"product_amount": amount,
}
self._do_post_request("stock/shoppinglist/remove-product", data)
def get_product_groups(self, query_filters: List[str] = None) -> List[LocationData]:
parsed_json = self._do_get_request("objects/product_groups", query_filters)
if parsed_json:
return [LocationData(**response) for response in parsed_json]
return []
def upload_product_picture(self, product_id: int, pic_path: str):
b64fn = base64.b64encode("{}.jpg".format(product_id).encode("ascii"))
req_url = "files/productpictures/" + str(b64fn, "utf-8")
with open(pic_path, "rb") as pic:
self._do_put_request(req_url, pic)
def update_product_pic(self, product_id: int):
pic_name = f"{product_id}.jpg"
data = {"picture_file_name": pic_name}
self._do_put_request(f"objects/products/{product_id}", data)
def get_userfields(self, entity: str, object_id: int):
url = f"userfields/{entity}/{object_id}"
return self._do_get_request(url)
def set_userfields(self, entity: str, object_id: int, key: str, value):
data = {key: value}
self._do_put_request(f"userfields/{entity}/{object_id}", data)
def get_last_db_changed(self):
resp = self._do_get_request("system/db-changed-time")
last_change_timestamp = parse_date(resp.get("changed_time"))
return last_change_timestamp
def get_system_info(self) -> SystemInfoDto:
parsed_json = self._do_get_request("system/info")
if parsed_json:
return SystemInfoDto(**parsed_json)
def get_system_time(self) -> SystemTimeDto:
parsed_json = self._do_get_request("system/time")
if parsed_json:
return SystemTimeDto(**parsed_json)
def get_system_config(self) -> SystemConfigDto:
parsed_json = self._do_get_request("system/config")
if parsed_json:
return SystemConfigDto(**parsed_json)
def get_tasks(self, query_filters: List[str] = None) -> List[TaskResponse]:
parsed_json = self._do_get_request("tasks", query_filters)
if parsed_json:
return [TaskResponse(**data) for data in parsed_json]
return []
def get_task(self, task_id: int) -> TaskResponse:
url = f"objects/tasks/{task_id}"
parsed_json = self._do_get_request(url)
return TaskResponse(**parsed_json)
def complete_task(self, task_id: int, done_time: datetime = None):
url = f"tasks/{task_id}/complete"
if done_time is None:
done_time = datetime.now()
localized_done_time = localize_datetime(done_time)
data = {"done_time": grocy_datetime_str(localized_done_time)}
self._do_post_request(url, data)
def get_meal_plan(self, query_filters: List[str] = None) -> List[MealPlanResponse]:
parsed_json = self._do_get_request("objects/meal_plan", query_filters)
if parsed_json:
return [MealPlanResponse(**data) for data in parsed_json]
return []
def get_recipe(self, object_id: int) -> RecipeDetailsResponse:
parsed_json = self._do_get_request(f"objects/recipes/{object_id}")
if parsed_json:
return RecipeDetailsResponse(**parsed_json)
def get_batteries(
self, query_filters: List[str] = None
) -> List[CurrentBatteryResponse]:
parsed_json = self._do_get_request("batteries", query_filters)
if parsed_json:
return [CurrentBatteryResponse(**data) for data in parsed_json]
return []
def get_battery(self, battery_id: int) -> BatteryDetailsResponse:
parsed_json = self._do_get_request(f"batteries/{battery_id}")
if parsed_json:
return BatteryDetailsResponse(**parsed_json)
def charge_battery(self, battery_id: int, tracked_time: datetime = None):
if tracked_time is None:
tracked_time = datetime.now()
localized_tracked_time = localize_datetime(tracked_time)
data = {"tracked_time": grocy_datetime_str(localized_tracked_time)}
return self._do_post_request(f"batteries/{battery_id}/charge", data)
def add_generic(self, entity_type: str, data):
return self._do_post_request(f"objects/{entity_type}", data)
def update_generic(self, entity_type: str, object_id: int, data):
return self._do_put_request(f"objects/{entity_type}/{object_id}", data)
def delete_generic(self, entity_type: str, object_id: int):
return self._do_delete_request(f"objects/{entity_type}/{object_id}")
def get_generic_objects_for_type(
self, entity_type: str, query_filters: List[str] = None
):
return self._do_get_request(f"objects/{entity_type}", query_filters)
def get_meal_plan_sections(
self, query_filters: List[str] = None
) -> List[MealPlanSectionResponse]:
parsed_json = self.get_generic_objects_for_type(
EntityType.MEAL_PLAN_SECTIONS, query_filters
)
if parsed_json:
return [MealPlanSectionResponse(**resp) for resp in parsed_json]
return []
def get_meal_plan_section(self, meal_plan_section_id) -> MealPlanSectionResponse:
parsed_json = self._do_get_request(
f"objects/meal_plan_sections?query%5B%5D=id%3D{meal_plan_section_id}"
)
if parsed_json and len(parsed_json) == 1:
return MealPlanSectionResponse(**parsed_json[0])
def get_users(self) -> List[UserDto]:
parsed_json = self._do_get_request("users")
if parsed_json:
return [UserDto(**user) for user in parsed_json]
return []
def get_user(self, user_id: int) -> UserDto:
query_params = []
if user_id:
query_params.append(f"id={user_id}")
parsed_json = self._do_get_request("users")
if parsed_json:
return UserDto(**parsed_json[0])
Classes
class BatteryData (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class BatteryData(BaseModel): id: int name: str description: str used_in: str charge_interval_days: int created_timestamp: datetime = Field(alias="row_created_timestamp") userfields: Optional[Dict] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var charge_interval_days : int
var created_timestamp : datetime.datetime
var description : str
var id : int
var name : str
var used_in : str
var userfields : Union[Dict, NoneType]
class BatteryDetailsResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class BatteryDetailsResponse(BaseModel): battery: BatteryData charge_cycles_count: int last_charged: Optional[datetime] = None last_tracked_time: Optional[datetime] = None next_estimated_charge_time: Optional[datetime] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var battery : BatteryData
var charge_cycles_count : int
var last_charged : Union[datetime.datetime, NoneType]
var last_tracked_time : Union[datetime.datetime, NoneType]
var next_estimated_charge_time : Union[datetime.datetime, NoneType]
class ChoreData (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class ChoreData(BaseModel): id: int name: str description: Optional[str] = None period_type: str period_config: Optional[str] = None period_days: Optional[int] = 0 track_date_only: bool rollover: bool assignment_type: Optional[str] = None assignment_config: Optional[str] = None next_execution_assigned_to_user_id: Optional[int] = None userfields: Optional[Dict] next_execution_assigned_to_user_id_validator = _field_not_empty_validator( "next_execution_assigned_to_user_id" )
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var assignment_config : Union[str, NoneType]
var assignment_type : Union[str, NoneType]
var description : Union[str, NoneType]
var id : int
var name : str
var next_execution_assigned_to_user_id : Union[int, NoneType]
var period_config : Union[str, NoneType]
var period_days : Union[int, NoneType]
var period_type : str
var rollover : bool
var track_date_only : bool
var userfields : Union[Dict, NoneType]
Static methods
def next_execution_assigned_to_user_id_validator()
-
Expand source code
def _none_if_empty_str(value: Any): if isinstance(value, str) and value == "": return None return value
class ChoreDetailsResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class ChoreDetailsResponse(BaseModel): chore: ChoreData last_tracked: Optional[datetime] = None next_estimated_execution_time: Optional[datetime] = None track_count: int = 0 next_execution_assigned_user: Optional[UserDto] = None last_done_by: Optional[UserDto] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var chore : ChoreData
var last_done_by : Union[UserDto, NoneType]
var last_tracked : Union[datetime.datetime, NoneType]
var next_estimated_execution_time : Union[datetime.datetime, NoneType]
var next_execution_assigned_user : Union[UserDto, NoneType]
var track_count : int
class CurrentBatteryResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class CurrentBatteryResponse(BaseModel): id: int last_tracked_time: Optional[datetime] = None next_estimated_charge_time: Optional[datetime] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var id : int
var last_tracked_time : Union[datetime.datetime, NoneType]
var next_estimated_charge_time : Union[datetime.datetime, NoneType]
class CurrentChoreResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class CurrentChoreResponse(BaseModel): chore_id: int last_tracked_time: Optional[datetime] = None next_estimated_execution_time: Optional[datetime] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var chore_id : int
var last_tracked_time : Union[datetime.datetime, NoneType]
var next_estimated_execution_time : Union[datetime.datetime, NoneType]
class CurrentStockResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class CurrentStockResponse(BaseModel): product_id: int amount: float best_before_date: date amount_opened: float amount_aggregated: float amount_opened_aggregated: float is_aggregated_amount: bool product: ProductData
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var amount : float
var amount_aggregated : float
var amount_opened : float
var amount_opened_aggregated : float
var best_before_date : datetime.date
var is_aggregated_amount : bool
var product : ProductData
var product_id : int
class CurrentVolatilStockResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class CurrentVolatilStockResponse(BaseModel): due_products: Optional[List[CurrentStockResponse]] = None overdue_products: Optional[List[CurrentStockResponse]] = None expired_products: Optional[List[CurrentStockResponse]] = None missing_products: Optional[List[MissingProductResponse]] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var due_products : Union[List[CurrentStockResponse], NoneType]
var expired_products : Union[List[CurrentStockResponse], NoneType]
var missing_products : Union[List[MissingProductResponse], NoneType]
var overdue_products : Union[List[CurrentStockResponse], NoneType]
class GrocyApiClient (base_url, api_key, port: int = 9192, path: str = None, verify_ssl=True, debug=False)
-
Expand source code
class GrocyApiClient(object): def __init__( self, base_url, api_key, port: int = DEFAULT_PORT_NUMBER, path: str = None, verify_ssl=True, debug=False, ): if debug: _enable_debug_mode() if path: self._base_url = "{}:{}/{}/api/".format(base_url, port, path) else: self._base_url = "{}:{}/api/".format(base_url, port) _LOGGER.debug(f"generated base url: {self._base_url}") self._api_key = api_key self._verify_ssl = verify_ssl if self._api_key == "demo_mode": self._headers = {"accept": "application/json"} else: self._headers = {"accept": "application/json", "GROCY-API-KEY": api_key} def _do_get_request(self, end_url: str, query_filters: List[str] = None): req_url = urljoin(self._base_url, end_url) params = None if query_filters: params = {"query[]": query_filters} resp = requests.get( req_url, verify=self._verify_ssl, headers=self._headers, params=params ) _LOGGER.debug("-->\tGET /%s", end_url) _LOGGER.debug("<--\t%d for /%s", resp.status_code, end_url) _LOGGER.debug("\t\t%s", resp.content) if resp.status_code >= 400: raise GrocyError(resp) if len(resp.content) > 0: return resp.json() def _do_post_request(self, end_url: str, data: dict): req_url = urljoin(self._base_url, end_url) resp = requests.post( req_url, verify=self._verify_ssl, headers=self._headers, json=data ) _LOGGER.debug("-->\tPOST /%s", end_url) _LOGGER.debug("\t\t%s", data) _LOGGER.debug("<--\t%d for /%s", resp.status_code, end_url) _LOGGER.debug("\t\t%s", resp.content) if resp.status_code >= 400: raise GrocyError(resp) if len(resp.content) > 0: return resp.json() def _do_put_request(self, end_url: str, data): req_url = urljoin(self._base_url, end_url) up_header = self._headers.copy() up_header["accept"] = "*/*" if isinstance(data, dict): up_header["Content-Type"] = "application/json" data = json.dumps(data) else: up_header["Content-Type"] = "application/octet-stream" resp = requests.put( req_url, verify=self._verify_ssl, headers=up_header, data=data ) _LOGGER.debug("-->\tPUT /%s", end_url) _LOGGER.debug("\t\t%s", data) _LOGGER.debug("<--\t%d for /%s", resp.status_code, end_url) _LOGGER.debug("\t\t%s", resp.content) if resp.status_code >= 400: raise GrocyError(resp) if len(resp.content) > 0: return resp.json() def _do_delete_request(self, end_url: str): req_url = urljoin(self._base_url, end_url) resp = requests.delete(req_url, verify=self._verify_ssl, headers=self._headers) _LOGGER.debug("-->\tDELETE /%s", end_url) _LOGGER.debug("<--\t%d for /%s", resp.status_code, end_url) _LOGGER.debug("\t\t%s", resp.content) if resp.status_code >= 400: raise GrocyError(resp) if len(resp.content) > 0: return resp.json() def get_stock(self) -> List[CurrentStockResponse]: parsed_json = self._do_get_request("stock") if parsed_json: return [CurrentStockResponse(**response) for response in parsed_json] return [] def get_volatile_stock(self) -> CurrentVolatilStockResponse: parsed_json = self._do_get_request("stock/volatile") return CurrentVolatilStockResponse(**parsed_json) def get_product(self, product_id) -> ProductDetailsResponse: url = f"stock/products/{product_id}" parsed_json = self._do_get_request(url) if parsed_json: return ProductDetailsResponse(**parsed_json) def get_product_by_barcode(self, barcode) -> ProductDetailsResponse: url = f"stock/products/by-barcode/{barcode}" parsed_json = self._do_get_request(url) if parsed_json: return ProductDetailsResponse(**parsed_json) def get_chores(self, query_filters: List[str] = None) -> List[CurrentChoreResponse]: parsed_json = self._do_get_request("chores", query_filters) if parsed_json: return [CurrentChoreResponse(**chore) for chore in parsed_json] return [] def get_chore(self, chore_id: int) -> ChoreDetailsResponse: url = f"chores/{chore_id}" parsed_json = self._do_get_request(url) if parsed_json: return ChoreDetailsResponse(**parsed_json) def execute_chore( self, chore_id: int, done_by: int = None, tracked_time: datetime = None, skipped: bool = False, ): if tracked_time is None: tracked_time = datetime.now() localized_tracked_time = localize_datetime(tracked_time) data = { "tracked_time": grocy_datetime_str(localized_tracked_time), "skipped": skipped, } if done_by is not None: data["done_by"] = done_by return self._do_post_request(f"chores/{chore_id}/execute", data) def add_product( self, product_id, amount: float, price: float, best_before_date: datetime = None, transaction_type: TransactionType = TransactionType.PURCHASE, ): data = { "amount": amount, "transaction_type": transaction_type.value, "price": price, } if best_before_date is not None: data["best_before_date"] = best_before_date.strftime("%Y-%m-%d") return self._do_post_request(f"stock/products/{product_id}/add", data) def consume_product( self, product_id: int, amount: float = 1, spoiled: bool = False, transaction_type: TransactionType = TransactionType.CONSUME, allow_subproduct_substitution: bool = False, ): data = { "amount": amount, "spoiled": spoiled, "transaction_type": transaction_type.value, "allow_subproduct_substitution": allow_subproduct_substitution, } self._do_post_request(f"stock/products/{product_id}/consume", data) def open_product( self, product_id: int, amount: float = 1, allow_subproduct_substitution: bool = False, ): data = { "amount": amount, "allow_subproduct_substitution": allow_subproduct_substitution, } self._do_post_request(f"stock/products/{product_id}/open", data) def consume_recipe( self, recipe_id: int, ): self._do_post_request(f"recipes/{recipe_id}/consume", None) def inventory_product( self, product_id: int, new_amount: float, best_before_date: datetime = None, shopping_location_id: int = None, location_id: int = None, price: float = None, ): data = { "new_amount": new_amount, } if best_before_date is not None: data["best_before_date"] = localize_datetime(best_before_date).strftime( "%Y-%m-%d" ) if shopping_location_id is not None: data["shopping_location_id"] = shopping_location_id if location_id is not None: data["location_id"] = location_id if price is not None: data["price"] = price parsed_json = self._do_post_request( f"stock/products/{product_id}/inventory", data ) if parsed_json: stockLog = [StockLogResponse(**response) for response in parsed_json] return stockLog[0] def add_product_by_barcode( self, barcode: str, amount: float, price: float, best_before_date: datetime = None, ) -> StockLogResponse: data = { "amount": amount, "transaction_type": TransactionType.PURCHASE.value, "price": price, } if best_before_date is not None: data["best_before_date"] = localize_datetime(best_before_date).strftime( "%Y-%m-%d" ) parsed_json = self._do_post_request( f"stock/products/by-barcode/{barcode}/add", data ) if parsed_json: stockLog = [StockLogResponse(**response) for response in parsed_json] return stockLog[0] def consume_product_by_barcode( self, barcode: str, amount: float = 1, spoiled: bool = False ): data = { "amount": amount, "spoiled": spoiled, "transaction_type": TransactionType.CONSUME.value, } parsed_json = self._do_post_request( f"stock/products/by-barcode/{barcode}/consume", data ) if parsed_json: stockLog = [StockLogResponse(**response) for response in parsed_json] return stockLog[0] def inventory_product_by_barcode( self, barcode: str, new_amount: float, best_before_date: datetime = None, location_id: int = None, price: float = None, ): data = { "new_amount": new_amount, } if best_before_date is not None: data["best_before_date"] = localize_datetime(best_before_date).strftime( "%Y-%m-%d" ) if location_id is not None: data["location_id"] = location_id if price is not None: data["price"] = price parsed_json = self._do_post_request( f"stock/products/by-barcode/{barcode}/inventory", data ) if parsed_json: stockLog = [StockLogResponse(**response) for response in parsed_json] return stockLog[0] def get_shopping_list( self, query_filters: List[str] = None ) -> List[ShoppingListItem]: parsed_json = self._do_get_request("objects/shopping_list", query_filters) if parsed_json: return [ShoppingListItem(**response) for response in parsed_json] return [] def add_missing_product_to_shopping_list(self, shopping_list_id: int = None): data = None if shopping_list_id: data = {"list_id": shopping_list_id} self._do_post_request("stock/shoppinglist/add-missing-products", data) def add_product_to_shopping_list( self, product_id: int, shopping_list_id: int = 1, amount: float = 1, quantity_unit_id: int = None, ): data = { "product_id": product_id, "list_id": shopping_list_id, "product_amount": amount, } if quantity_unit_id: data["qu_id"] = quantity_unit_id self._do_post_request("stock/shoppinglist/add-product", data) def clear_shopping_list(self, shopping_list_id: int = 1): data = {"list_id": shopping_list_id} self._do_post_request("stock/shoppinglist/clear", data) def remove_product_in_shopping_list( self, product_id: int, shopping_list_id: int = 1, amount: float = 1 ): data = { "product_id": product_id, "list_id": shopping_list_id, "product_amount": amount, } self._do_post_request("stock/shoppinglist/remove-product", data) def get_product_groups(self, query_filters: List[str] = None) -> List[LocationData]: parsed_json = self._do_get_request("objects/product_groups", query_filters) if parsed_json: return [LocationData(**response) for response in parsed_json] return [] def upload_product_picture(self, product_id: int, pic_path: str): b64fn = base64.b64encode("{}.jpg".format(product_id).encode("ascii")) req_url = "files/productpictures/" + str(b64fn, "utf-8") with open(pic_path, "rb") as pic: self._do_put_request(req_url, pic) def update_product_pic(self, product_id: int): pic_name = f"{product_id}.jpg" data = {"picture_file_name": pic_name} self._do_put_request(f"objects/products/{product_id}", data) def get_userfields(self, entity: str, object_id: int): url = f"userfields/{entity}/{object_id}" return self._do_get_request(url) def set_userfields(self, entity: str, object_id: int, key: str, value): data = {key: value} self._do_put_request(f"userfields/{entity}/{object_id}", data) def get_last_db_changed(self): resp = self._do_get_request("system/db-changed-time") last_change_timestamp = parse_date(resp.get("changed_time")) return last_change_timestamp def get_system_info(self) -> SystemInfoDto: parsed_json = self._do_get_request("system/info") if parsed_json: return SystemInfoDto(**parsed_json) def get_system_time(self) -> SystemTimeDto: parsed_json = self._do_get_request("system/time") if parsed_json: return SystemTimeDto(**parsed_json) def get_system_config(self) -> SystemConfigDto: parsed_json = self._do_get_request("system/config") if parsed_json: return SystemConfigDto(**parsed_json) def get_tasks(self, query_filters: List[str] = None) -> List[TaskResponse]: parsed_json = self._do_get_request("tasks", query_filters) if parsed_json: return [TaskResponse(**data) for data in parsed_json] return [] def get_task(self, task_id: int) -> TaskResponse: url = f"objects/tasks/{task_id}" parsed_json = self._do_get_request(url) return TaskResponse(**parsed_json) def complete_task(self, task_id: int, done_time: datetime = None): url = f"tasks/{task_id}/complete" if done_time is None: done_time = datetime.now() localized_done_time = localize_datetime(done_time) data = {"done_time": grocy_datetime_str(localized_done_time)} self._do_post_request(url, data) def get_meal_plan(self, query_filters: List[str] = None) -> List[MealPlanResponse]: parsed_json = self._do_get_request("objects/meal_plan", query_filters) if parsed_json: return [MealPlanResponse(**data) for data in parsed_json] return [] def get_recipe(self, object_id: int) -> RecipeDetailsResponse: parsed_json = self._do_get_request(f"objects/recipes/{object_id}") if parsed_json: return RecipeDetailsResponse(**parsed_json) def get_batteries( self, query_filters: List[str] = None ) -> List[CurrentBatteryResponse]: parsed_json = self._do_get_request("batteries", query_filters) if parsed_json: return [CurrentBatteryResponse(**data) for data in parsed_json] return [] def get_battery(self, battery_id: int) -> BatteryDetailsResponse: parsed_json = self._do_get_request(f"batteries/{battery_id}") if parsed_json: return BatteryDetailsResponse(**parsed_json) def charge_battery(self, battery_id: int, tracked_time: datetime = None): if tracked_time is None: tracked_time = datetime.now() localized_tracked_time = localize_datetime(tracked_time) data = {"tracked_time": grocy_datetime_str(localized_tracked_time)} return self._do_post_request(f"batteries/{battery_id}/charge", data) def add_generic(self, entity_type: str, data): return self._do_post_request(f"objects/{entity_type}", data) def update_generic(self, entity_type: str, object_id: int, data): return self._do_put_request(f"objects/{entity_type}/{object_id}", data) def delete_generic(self, entity_type: str, object_id: int): return self._do_delete_request(f"objects/{entity_type}/{object_id}") def get_generic_objects_for_type( self, entity_type: str, query_filters: List[str] = None ): return self._do_get_request(f"objects/{entity_type}", query_filters) def get_meal_plan_sections( self, query_filters: List[str] = None ) -> List[MealPlanSectionResponse]: parsed_json = self.get_generic_objects_for_type( EntityType.MEAL_PLAN_SECTIONS, query_filters ) if parsed_json: return [MealPlanSectionResponse(**resp) for resp in parsed_json] return [] def get_meal_plan_section(self, meal_plan_section_id) -> MealPlanSectionResponse: parsed_json = self._do_get_request( f"objects/meal_plan_sections?query%5B%5D=id%3D{meal_plan_section_id}" ) if parsed_json and len(parsed_json) == 1: return MealPlanSectionResponse(**parsed_json[0]) def get_users(self) -> List[UserDto]: parsed_json = self._do_get_request("users") if parsed_json: return [UserDto(**user) for user in parsed_json] return [] def get_user(self, user_id: int) -> UserDto: query_params = [] if user_id: query_params.append(f"id={user_id}") parsed_json = self._do_get_request("users") if parsed_json: return UserDto(**parsed_json[0])
Methods
def add_generic(self, entity_type: str, data)
-
Expand source code
def add_generic(self, entity_type: str, data): return self._do_post_request(f"objects/{entity_type}", data)
def add_missing_product_to_shopping_list(self, shopping_list_id: int = None)
-
Expand source code
def add_missing_product_to_shopping_list(self, shopping_list_id: int = None): data = None if shopping_list_id: data = {"list_id": shopping_list_id} self._do_post_request("stock/shoppinglist/add-missing-products", data)
def add_product(self, product_id, amount: float, price: float, best_before_date: datetime.datetime = None, transaction_type: TransactionType = TransactionType.PURCHASE)
-
Expand source code
def add_product( self, product_id, amount: float, price: float, best_before_date: datetime = None, transaction_type: TransactionType = TransactionType.PURCHASE, ): data = { "amount": amount, "transaction_type": transaction_type.value, "price": price, } if best_before_date is not None: data["best_before_date"] = best_before_date.strftime("%Y-%m-%d") return self._do_post_request(f"stock/products/{product_id}/add", data)
def add_product_by_barcode(self, barcode: str, amount: float, price: float, best_before_date: datetime.datetime = None) ‑> StockLogResponse
-
Expand source code
def add_product_by_barcode( self, barcode: str, amount: float, price: float, best_before_date: datetime = None, ) -> StockLogResponse: data = { "amount": amount, "transaction_type": TransactionType.PURCHASE.value, "price": price, } if best_before_date is not None: data["best_before_date"] = localize_datetime(best_before_date).strftime( "%Y-%m-%d" ) parsed_json = self._do_post_request( f"stock/products/by-barcode/{barcode}/add", data ) if parsed_json: stockLog = [StockLogResponse(**response) for response in parsed_json] return stockLog[0]
def add_product_to_shopping_list(self, product_id: int, shopping_list_id: int = 1, amount: float = 1, quantity_unit_id: int = None)
-
Expand source code
def add_product_to_shopping_list( self, product_id: int, shopping_list_id: int = 1, amount: float = 1, quantity_unit_id: int = None, ): data = { "product_id": product_id, "list_id": shopping_list_id, "product_amount": amount, } if quantity_unit_id: data["qu_id"] = quantity_unit_id self._do_post_request("stock/shoppinglist/add-product", data)
def charge_battery(self, battery_id: int, tracked_time: datetime.datetime = None)
-
Expand source code
def charge_battery(self, battery_id: int, tracked_time: datetime = None): if tracked_time is None: tracked_time = datetime.now() localized_tracked_time = localize_datetime(tracked_time) data = {"tracked_time": grocy_datetime_str(localized_tracked_time)} return self._do_post_request(f"batteries/{battery_id}/charge", data)
def clear_shopping_list(self, shopping_list_id: int = 1)
-
Expand source code
def clear_shopping_list(self, shopping_list_id: int = 1): data = {"list_id": shopping_list_id} self._do_post_request("stock/shoppinglist/clear", data)
def complete_task(self, task_id: int, done_time: datetime.datetime = None)
-
Expand source code
def complete_task(self, task_id: int, done_time: datetime = None): url = f"tasks/{task_id}/complete" if done_time is None: done_time = datetime.now() localized_done_time = localize_datetime(done_time) data = {"done_time": grocy_datetime_str(localized_done_time)} self._do_post_request(url, data)
def consume_product(self, product_id: int, amount: float = 1, spoiled: bool = False, transaction_type: TransactionType = TransactionType.CONSUME, allow_subproduct_substitution: bool = False)
-
Expand source code
def consume_product( self, product_id: int, amount: float = 1, spoiled: bool = False, transaction_type: TransactionType = TransactionType.CONSUME, allow_subproduct_substitution: bool = False, ): data = { "amount": amount, "spoiled": spoiled, "transaction_type": transaction_type.value, "allow_subproduct_substitution": allow_subproduct_substitution, } self._do_post_request(f"stock/products/{product_id}/consume", data)
def consume_product_by_barcode(self, barcode: str, amount: float = 1, spoiled: bool = False)
-
Expand source code
def consume_product_by_barcode( self, barcode: str, amount: float = 1, spoiled: bool = False ): data = { "amount": amount, "spoiled": spoiled, "transaction_type": TransactionType.CONSUME.value, } parsed_json = self._do_post_request( f"stock/products/by-barcode/{barcode}/consume", data ) if parsed_json: stockLog = [StockLogResponse(**response) for response in parsed_json] return stockLog[0]
def consume_recipe(self, recipe_id: int)
-
Expand source code
def consume_recipe( self, recipe_id: int, ): self._do_post_request(f"recipes/{recipe_id}/consume", None)
def delete_generic(self, entity_type: str, object_id: int)
-
Expand source code
def delete_generic(self, entity_type: str, object_id: int): return self._do_delete_request(f"objects/{entity_type}/{object_id}")
def execute_chore(self, chore_id: int, done_by: int = None, tracked_time: datetime.datetime = None, skipped: bool = False)
-
Expand source code
def execute_chore( self, chore_id: int, done_by: int = None, tracked_time: datetime = None, skipped: bool = False, ): if tracked_time is None: tracked_time = datetime.now() localized_tracked_time = localize_datetime(tracked_time) data = { "tracked_time": grocy_datetime_str(localized_tracked_time), "skipped": skipped, } if done_by is not None: data["done_by"] = done_by return self._do_post_request(f"chores/{chore_id}/execute", data)
def get_batteries(self, query_filters: List[str] = None) ‑> List[CurrentBatteryResponse]
-
Expand source code
def get_batteries( self, query_filters: List[str] = None ) -> List[CurrentBatteryResponse]: parsed_json = self._do_get_request("batteries", query_filters) if parsed_json: return [CurrentBatteryResponse(**data) for data in parsed_json] return []
def get_battery(self, battery_id: int) ‑> BatteryDetailsResponse
-
Expand source code
def get_battery(self, battery_id: int) -> BatteryDetailsResponse: parsed_json = self._do_get_request(f"batteries/{battery_id}") if parsed_json: return BatteryDetailsResponse(**parsed_json)
def get_chore(self, chore_id: int) ‑> ChoreDetailsResponse
-
Expand source code
def get_chore(self, chore_id: int) -> ChoreDetailsResponse: url = f"chores/{chore_id}" parsed_json = self._do_get_request(url) if parsed_json: return ChoreDetailsResponse(**parsed_json)
def get_chores(self, query_filters: List[str] = None) ‑> List[CurrentChoreResponse]
-
Expand source code
def get_chores(self, query_filters: List[str] = None) -> List[CurrentChoreResponse]: parsed_json = self._do_get_request("chores", query_filters) if parsed_json: return [CurrentChoreResponse(**chore) for chore in parsed_json] return []
def get_generic_objects_for_type(self, entity_type: str, query_filters: List[str] = None)
-
Expand source code
def get_generic_objects_for_type( self, entity_type: str, query_filters: List[str] = None ): return self._do_get_request(f"objects/{entity_type}", query_filters)
def get_last_db_changed(self)
-
Expand source code
def get_last_db_changed(self): resp = self._do_get_request("system/db-changed-time") last_change_timestamp = parse_date(resp.get("changed_time")) return last_change_timestamp
def get_meal_plan(self, query_filters: List[str] = None) ‑> List[MealPlanResponse]
-
Expand source code
def get_meal_plan(self, query_filters: List[str] = None) -> List[MealPlanResponse]: parsed_json = self._do_get_request("objects/meal_plan", query_filters) if parsed_json: return [MealPlanResponse(**data) for data in parsed_json] return []
def get_meal_plan_section(self, meal_plan_section_id) ‑> MealPlanSectionResponse
-
Expand source code
def get_meal_plan_section(self, meal_plan_section_id) -> MealPlanSectionResponse: parsed_json = self._do_get_request( f"objects/meal_plan_sections?query%5B%5D=id%3D{meal_plan_section_id}" ) if parsed_json and len(parsed_json) == 1: return MealPlanSectionResponse(**parsed_json[0])
def get_meal_plan_sections(self, query_filters: List[str] = None) ‑> List[MealPlanSectionResponse]
-
Expand source code
def get_meal_plan_sections( self, query_filters: List[str] = None ) -> List[MealPlanSectionResponse]: parsed_json = self.get_generic_objects_for_type( EntityType.MEAL_PLAN_SECTIONS, query_filters ) if parsed_json: return [MealPlanSectionResponse(**resp) for resp in parsed_json] return []
def get_product(self, product_id) ‑> ProductDetailsResponse
-
Expand source code
def get_product(self, product_id) -> ProductDetailsResponse: url = f"stock/products/{product_id}" parsed_json = self._do_get_request(url) if parsed_json: return ProductDetailsResponse(**parsed_json)
def get_product_by_barcode(self, barcode) ‑> ProductDetailsResponse
-
Expand source code
def get_product_by_barcode(self, barcode) -> ProductDetailsResponse: url = f"stock/products/by-barcode/{barcode}" parsed_json = self._do_get_request(url) if parsed_json: return ProductDetailsResponse(**parsed_json)
def get_product_groups(self, query_filters: List[str] = None) ‑> List[LocationData]
-
Expand source code
def get_product_groups(self, query_filters: List[str] = None) -> List[LocationData]: parsed_json = self._do_get_request("objects/product_groups", query_filters) if parsed_json: return [LocationData(**response) for response in parsed_json] return []
def get_recipe(self, object_id: int) ‑> RecipeDetailsResponse
-
Expand source code
def get_recipe(self, object_id: int) -> RecipeDetailsResponse: parsed_json = self._do_get_request(f"objects/recipes/{object_id}") if parsed_json: return RecipeDetailsResponse(**parsed_json)
def get_shopping_list(self, query_filters: List[str] = None) ‑> List[ShoppingListItem]
-
Expand source code
def get_shopping_list( self, query_filters: List[str] = None ) -> List[ShoppingListItem]: parsed_json = self._do_get_request("objects/shopping_list", query_filters) if parsed_json: return [ShoppingListItem(**response) for response in parsed_json] return []
def get_stock(self) ‑> List[CurrentStockResponse]
-
Expand source code
def get_stock(self) -> List[CurrentStockResponse]: parsed_json = self._do_get_request("stock") if parsed_json: return [CurrentStockResponse(**response) for response in parsed_json] return []
def get_system_config(self) ‑> SystemConfigDto
-
Expand source code
def get_system_config(self) -> SystemConfigDto: parsed_json = self._do_get_request("system/config") if parsed_json: return SystemConfigDto(**parsed_json)
def get_system_info(self) ‑> SystemInfoDto
-
Expand source code
def get_system_info(self) -> SystemInfoDto: parsed_json = self._do_get_request("system/info") if parsed_json: return SystemInfoDto(**parsed_json)
def get_system_time(self) ‑> SystemTimeDto
-
Expand source code
def get_system_time(self) -> SystemTimeDto: parsed_json = self._do_get_request("system/time") if parsed_json: return SystemTimeDto(**parsed_json)
def get_task(self, task_id: int) ‑> TaskResponse
-
Expand source code
def get_task(self, task_id: int) -> TaskResponse: url = f"objects/tasks/{task_id}" parsed_json = self._do_get_request(url) return TaskResponse(**parsed_json)
def get_tasks(self, query_filters: List[str] = None) ‑> List[TaskResponse]
-
Expand source code
def get_tasks(self, query_filters: List[str] = None) -> List[TaskResponse]: parsed_json = self._do_get_request("tasks", query_filters) if parsed_json: return [TaskResponse(**data) for data in parsed_json] return []
def get_user(self, user_id: int) ‑> UserDto
-
Expand source code
def get_user(self, user_id: int) -> UserDto: query_params = [] if user_id: query_params.append(f"id={user_id}") parsed_json = self._do_get_request("users") if parsed_json: return UserDto(**parsed_json[0])
def get_userfields(self, entity: str, object_id: int)
-
Expand source code
def get_userfields(self, entity: str, object_id: int): url = f"userfields/{entity}/{object_id}" return self._do_get_request(url)
def get_users(self) ‑> List[UserDto]
-
Expand source code
def get_users(self) -> List[UserDto]: parsed_json = self._do_get_request("users") if parsed_json: return [UserDto(**user) for user in parsed_json] return []
def get_volatile_stock(self) ‑> CurrentVolatilStockResponse
-
Expand source code
def get_volatile_stock(self) -> CurrentVolatilStockResponse: parsed_json = self._do_get_request("stock/volatile") return CurrentVolatilStockResponse(**parsed_json)
def inventory_product(self, product_id: int, new_amount: float, best_before_date: datetime.datetime = None, shopping_location_id: int = None, location_id: int = None, price: float = None)
-
Expand source code
def inventory_product( self, product_id: int, new_amount: float, best_before_date: datetime = None, shopping_location_id: int = None, location_id: int = None, price: float = None, ): data = { "new_amount": new_amount, } if best_before_date is not None: data["best_before_date"] = localize_datetime(best_before_date).strftime( "%Y-%m-%d" ) if shopping_location_id is not None: data["shopping_location_id"] = shopping_location_id if location_id is not None: data["location_id"] = location_id if price is not None: data["price"] = price parsed_json = self._do_post_request( f"stock/products/{product_id}/inventory", data ) if parsed_json: stockLog = [StockLogResponse(**response) for response in parsed_json] return stockLog[0]
def inventory_product_by_barcode(self, barcode: str, new_amount: float, best_before_date: datetime.datetime = None, location_id: int = None, price: float = None)
-
Expand source code
def inventory_product_by_barcode( self, barcode: str, new_amount: float, best_before_date: datetime = None, location_id: int = None, price: float = None, ): data = { "new_amount": new_amount, } if best_before_date is not None: data["best_before_date"] = localize_datetime(best_before_date).strftime( "%Y-%m-%d" ) if location_id is not None: data["location_id"] = location_id if price is not None: data["price"] = price parsed_json = self._do_post_request( f"stock/products/by-barcode/{barcode}/inventory", data ) if parsed_json: stockLog = [StockLogResponse(**response) for response in parsed_json] return stockLog[0]
def open_product(self, product_id: int, amount: float = 1, allow_subproduct_substitution: bool = False)
-
Expand source code
def open_product( self, product_id: int, amount: float = 1, allow_subproduct_substitution: bool = False, ): data = { "amount": amount, "allow_subproduct_substitution": allow_subproduct_substitution, } self._do_post_request(f"stock/products/{product_id}/open", data)
def remove_product_in_shopping_list(self, product_id: int, shopping_list_id: int = 1, amount: float = 1)
-
Expand source code
def remove_product_in_shopping_list( self, product_id: int, shopping_list_id: int = 1, amount: float = 1 ): data = { "product_id": product_id, "list_id": shopping_list_id, "product_amount": amount, } self._do_post_request("stock/shoppinglist/remove-product", data)
def set_userfields(self, entity: str, object_id: int, key: str, value)
-
Expand source code
def set_userfields(self, entity: str, object_id: int, key: str, value): data = {key: value} self._do_put_request(f"userfields/{entity}/{object_id}", data)
def update_generic(self, entity_type: str, object_id: int, data)
-
Expand source code
def update_generic(self, entity_type: str, object_id: int, data): return self._do_put_request(f"objects/{entity_type}/{object_id}", data)
def update_product_pic(self, product_id: int)
-
Expand source code
def update_product_pic(self, product_id: int): pic_name = f"{product_id}.jpg" data = {"picture_file_name": pic_name} self._do_put_request(f"objects/products/{product_id}", data)
def upload_product_picture(self, product_id: int, pic_path: str)
-
Expand source code
def upload_product_picture(self, product_id: int, pic_path: str): b64fn = base64.b64encode("{}.jpg".format(product_id).encode("ascii")) req_url = "files/productpictures/" + str(b64fn, "utf-8") with open(pic_path, "rb") as pic: self._do_put_request(req_url, pic)
class GrocyVersionDto (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class GrocyVersionDto(BaseModel): version: str = Field(alias="Version") release_date: date = Field(alias="ReleaseDate")
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var release_date : datetime.date
var version : str
class LocationData (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class LocationData(BaseModel): id: int name: str description: Optional[str] = None row_created_timestamp: datetime
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var description : Union[str, NoneType]
var id : int
var name : str
var row_created_timestamp : datetime.datetime
class MealPlanResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class MealPlanResponse(BaseModel): id: int day: date type: str recipe_id: Optional[int] = None recipe_servings: Optional[int] = None note: Optional[str] = None product_id: Optional[int] = None product_amount: Optional[float] = None product_qu_id: Optional[str] = None row_created_timestamp: datetime userfields: Optional[Dict] = None section_id: Optional[int] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var day : datetime.date
var id : int
var note : Union[str, NoneType]
var product_amount : Union[float, NoneType]
var product_id : Union[int, NoneType]
var product_qu_id : Union[str, NoneType]
var recipe_id : Union[int, NoneType]
var recipe_servings : Union[int, NoneType]
var row_created_timestamp : datetime.datetime
var section_id : Union[int, NoneType]
var type : str
var userfields : Union[Dict, NoneType]
class MealPlanSectionResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class MealPlanSectionResponse(BaseModel): id: Optional[int] = None name: str sort_number: Optional[int] = None row_created_timestamp: datetime sort_number_validator = _field_not_empty_validator("sort_number")
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var id : Union[int, NoneType]
var name : str
var row_created_timestamp : datetime.datetime
var sort_number : Union[int, NoneType]
Static methods
def sort_number_validator()
-
Expand source code
def _none_if_empty_str(value: Any): if isinstance(value, str) and value == "": return None return value
class MissingProductResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class MissingProductResponse(BaseModel): id: int name: str amount_missing: float is_partly_in_stock: bool
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var amount_missing : float
var id : int
var is_partly_in_stock : bool
var name : str
class ProductBarcodeData (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class ProductBarcodeData(BaseModel): barcode: str
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var barcode : str
class ProductData (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class ProductData(BaseModel): id: int name: str description: Optional[str] = None location_id: Optional[int] = None product_group_id: Optional[int] = None qu_id_stock: int qu_id_purchase: int picture_file_name: Optional[str] = None allow_partial_units_in_stock: Optional[bool] = False row_created_timestamp: datetime min_stock_amount: Optional[float] default_best_before_days: int location_id_validator = _field_not_empty_validator("location_id") product_group_id_validator = _field_not_empty_validator("product_group_id")
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var allow_partial_units_in_stock : Union[bool, NoneType]
var default_best_before_days : int
var description : Union[str, NoneType]
var id : int
var location_id : Union[int, NoneType]
var min_stock_amount : Union[float, NoneType]
var name : str
var picture_file_name : Union[str, NoneType]
var product_group_id : Union[int, NoneType]
var qu_id_purchase : int
var qu_id_stock : int
var row_created_timestamp : datetime.datetime
Static methods
def location_id_validator()
-
Expand source code
def _none_if_empty_str(value: Any): if isinstance(value, str) and value == "": return None return value
def product_group_id_validator()
-
Expand source code
def _none_if_empty_str(value: Any): if isinstance(value, str) and value == "": return None return value
class ProductDetailsResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class ProductDetailsResponse(BaseModel): last_purchased: Optional[date] = None last_used: Optional[date] = None stock_amount: float stock_amount_opened: float next_best_before_date: Optional[date] = None last_price: Optional[float] = None product: ProductData quantity_unit_stock: QuantityUnitData default_quantity_unit_purchase: QuantityUnitData barcodes: Optional[List[ProductBarcodeData]] = Field(alias="product_barcodes") location: Optional[LocationData] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var barcodes : Union[List[ProductBarcodeData], NoneType]
var default_quantity_unit_purchase : QuantityUnitData
var last_price : Union[float, NoneType]
var last_purchased : Union[datetime.date, NoneType]
var last_used : Union[datetime.date, NoneType]
var location : Union[LocationData, NoneType]
var next_best_before_date : Union[datetime.date, NoneType]
var product : ProductData
var quantity_unit_stock : QuantityUnitData
var stock_amount : float
var stock_amount_opened : float
class QuantityUnitData (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class QuantityUnitData(BaseModel): id: int name: str name_plural: str description: Optional[str] = None row_created_timestamp: datetime
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var description : Union[str, NoneType]
var id : int
var name : str
var name_plural : str
var row_created_timestamp : datetime.datetime
class RecipeDetailsResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class RecipeDetailsResponse(BaseModel): id: Optional[int] = None name: str description: Optional[str] = None base_servings: int desired_servings: int picture_file_name: Optional[str] row_created_timestamp: datetime userfields: Optional[Dict] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var base_servings : int
var description : Union[str, NoneType]
var desired_servings : int
var id : Union[int, NoneType]
var name : str
var picture_file_name : Union[str, NoneType]
var row_created_timestamp : datetime.datetime
var userfields : Union[Dict, NoneType]
class ShoppingListItem (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class ShoppingListItem(BaseModel): id: int product_id: Optional[int] = None note: Optional[str] = None amount: float row_created_timestamp: datetime shopping_list_id: int done: int
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var amount : float
var done : int
var id : int
var note : Union[str, NoneType]
var product_id : Union[int, NoneType]
var row_created_timestamp : datetime.datetime
var shopping_list_id : int
class StockLogResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class StockLogResponse(BaseModel): id: int product_id: int amount: float best_before_date: date purchased_date: date used_date: Optional[date] = None spoiled: bool = False stock_id: str transaction_id: str transaction_type: TransactionType
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var amount : float
var best_before_date : datetime.date
var id : int
var product_id : int
var purchased_date : datetime.date
var spoiled : bool
var stock_id : str
var transaction_id : str
var transaction_type : TransactionType
var used_date : Union[datetime.date, NoneType]
class SystemConfigDto (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class SystemConfigDto(BaseModel, extra=Extra.allow): username: str = Field(alias="USER_USERNAME") base_path: str = Field(alias="BASE_PATH") base_url: str = Field(alias="BASE_URL") mode: str = Field(alias="MODE") default_locale: str = Field(alias="DEFAULT_LOCALE") locale: str = Field(alias="LOCALE") currency: str = Field(alias="CURRENCY") feature_flags: Dict[str, Any] @root_validator(pre=True) def feature_flags_root_validator(cls, fields: Dict[str, Any]) -> Dict[str, Any]: """Pydantic root validator to add all "FEATURE_FLAG_" settings to Dict.""" features: Dict[str, Any] = {} class_defined_fields = cls.__fields__.values() for field, value in fields.items(): if field.startswith("FEATURE_FLAG_") and field not in class_defined_fields: features[field] = value fields["feature_flags"] = features return fields
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var base_path : str
var base_url : str
var currency : str
var default_locale : str
var feature_flags : Dict[str, Any]
var locale : str
var mode : str
var username : str
Static methods
def feature_flags_root_validator(fields: Dict[str, Any]) ‑> Dict[str, Any]
-
Pydantic root validator to add all "FEATURE_FLAG_" settings to Dict.
Expand source code
@root_validator(pre=True) def feature_flags_root_validator(cls, fields: Dict[str, Any]) -> Dict[str, Any]: """Pydantic root validator to add all "FEATURE_FLAG_" settings to Dict.""" features: Dict[str, Any] = {} class_defined_fields = cls.__fields__.values() for field, value in fields.items(): if field.startswith("FEATURE_FLAG_") and field not in class_defined_fields: features[field] = value fields["feature_flags"] = features return fields
class SystemInfoDto (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class SystemInfoDto(BaseModel): grocy_version_info: GrocyVersionDto = Field(alias="grocy_version") php_version: str sqlite_version: str os: str client: str
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var client : str
var grocy_version_info : GrocyVersionDto
var os : str
var php_version : str
var sqlite_version : str
class SystemTimeDto (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class SystemTimeDto(BaseModel): timezone: str time_local: datetime time_local_sqlite3: datetime time_utc: datetime timestamp: int
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var time_local : datetime.datetime
var time_local_sqlite3 : datetime.datetime
var time_utc : datetime.datetime
var timestamp : int
var timezone : str
class TaskCategoryDto (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class TaskCategoryDto(BaseModel): id: int name: str description: Optional[str] = None row_created_timestamp: datetime
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var description : Union[str, NoneType]
var id : int
var name : str
var row_created_timestamp : datetime.datetime
class TaskResponse (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class TaskResponse(BaseModel): id: int name: str description: Optional[str] = None due_date: Optional[date] = None done: int done_timestamp: Optional[datetime] = None category_id: Optional[int] = None category: Optional[TaskCategoryDto] = None assigned_to_user_id: Optional[int] = None assigned_to_user: Optional[UserDto] = None userfields: Optional[Dict] = None due_date_validator = _field_not_empty_validator("due_date") category_id_validator = _field_not_empty_validator("category_id") assigned_to_user_id_validator = _field_not_empty_validator("assigned_to_user_id")
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var assigned_to_user : Union[UserDto, NoneType]
var assigned_to_user_id : Union[int, NoneType]
var category : Union[TaskCategoryDto, NoneType]
var category_id : Union[int, NoneType]
var description : Union[str, NoneType]
var done : int
var done_timestamp : Union[datetime.datetime, NoneType]
var due_date : Union[datetime.date, NoneType]
var id : int
var name : str
var userfields : Union[Dict, NoneType]
Static methods
def assigned_to_user_id_validator()
-
Expand source code
def _none_if_empty_str(value: Any): if isinstance(value, str) and value == "": return None return value
def category_id_validator()
-
Expand source code
def _none_if_empty_str(value: Any): if isinstance(value, str) and value == "": return None return value
def due_date_validator()
-
Expand source code
def _none_if_empty_str(value: Any): if isinstance(value, str) and value == "": return None return value
class TransactionType (value, names=None, *, module=None, qualname=None, type=None, start=1)
-
An enumeration.
Expand source code
class TransactionType(Enum): PURCHASE = "purchase" CONSUME = "consume" INVENTORY_CORRECTION = "inventory-correction" PRODUCT_OPENED = "product-opened"
Ancestors
- enum.Enum
Class variables
var CONSUME
var INVENTORY_CORRECTION
var PRODUCT_OPENED
var PURCHASE
class UserDto (**data: Any)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class UserDto(BaseModel): id: int username: str first_name: Optional[str] = None last_name: Optional[str] = None display_name: Optional[str] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var display_name : Union[str, NoneType]
var first_name : Union[str, NoneType]
var id : int
var last_name : Union[str, NoneType]
var username : str