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 batteryBatteryData
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 choreChoreData
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 productProductData
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_purchaseQuantityUnitData
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 productProductData
var quantity_unit_stockQuantityUnitData
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_typeTransactionType
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_infoGrocyVersionDto
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