API Reference

API reference for the freshpointsync library.

The API is designed to provide both high-level interfaces for fetching and updating FreshPoint product data, as well as low-level interfaces that perform specific tasks, such as fetching a webpage or parsing HTML content.

Below is a list of the high-level classes and functions that are available straigh from freshpointsync.

• freshpointsync.ProductPage

• freshpointsync.ProductPageData

• freshpointsync.ProductPageHub

• freshpointsync.ProductPageHubData

• freshpointsync.Product

• freshpointsync.ProductUpdateEvent

• freshpointsync.is_valid_handler()

freshpointsync.client

The client package provides classes for fetching HTML content from a specific FreshPoint webpage. It is a part of the low-level API.

class ProductDataFetchClient(timeout=None, max_retries=5)

Bases: object

Asynchronous utility for fetching contents of a specified FreshPoint.cz web page.

This class wraps an aiohttp.ClientSession object and provides additional features like retries, timeouts, logging, and comprehensive error handling.

Parameters:

• timeout (ClientTimeout | int | float | None)

• max_retries (int)

BASE_URL = 'https://my.freshpoint.cz'

The base URL of the FreshPoint.cz website.

classmethod get_page_url(location_id)

Generate a page URL for a given location ID.

Parameters:

location_id (int) – The ID of the location, as it appears in the FreshPoint.cz web page URL. For example, in https://my.freshpoint.cz/device/product-list/296, the ID is 296.

Returns:

The full page URL for the given location ID.

Return type:

str

property session: ClientSession | None

The aiohttp client session object used for fetching data.

property is_session_closed: bool

Check if the client session is closed.

Returns:

True if the client session is closed, False otherwise.

Return type:

bool

async set_session(session)

Set the client session object. If the previous session is not closed, it is closed before setting the new one.

Parameters:

session (aiohttp.ClientSession) – The client session to set.

Return type:

None

property timeout: ClientTimeout

Client request timeout.

set_timeout(timeout)

Set the client request timeout.

Parameters:

timeout (Optional[Union[aiohttp.ClientTimeout, int, float]]) – The timeout value. It can be an aiohttp.ClientTimeout object, an integer or a float representing the total timeout in seconds, or None for the default aiohttp client timeout.

Raises:

ValueError – If the timeout value is negative or invalid.

Return type:

None

property max_retries: int

The maximum number of retries for fetching data.

set_max_retries(max_retries)

Set the maximum number of retries for the fetching data.

Parameters:

max_retries (int) – The maximum number of retries.

Raises:

• TypeError – If the max_retries value is not an integer.

• ValueError – If the max_retries value is less than 1.

Return type:

None

async start_session()

Start an aiohttp client session if one is not already started.

Return type:

None

async close_session()

Close the aiohttp client session if one is open.

Return type:

None

async fetch(location_id, timeout=None, max_retries=None)

Fetch HTML data from a FreshPoint.cz web page.

Parameters:

• location_id (Union[int, str]) – The ID the FreshPoint location.

• timeout (Optional[Union[aiohttp.ClientTimeout, int, float]]) – The timeout for the request. If None, the default timeout is used.

• max_retries (Optional[int]) – The maximum number of retries. If None, the default number of retries is used.

Returns:

The fetched data as a string, or None if the fetch failed.

Return type:

Optional[str]

freshpointsync.page

The page package provides classes for interacting with a FreshPoint webpage. Classes from this module are composed of classes from the client, parser, product, and update packages. They form the high-level API.

class FetchInfo(contents, contents_hash, is_updated)

Bases: NamedTuple

Named tuple for a product page fetch information.

Parameters:

• contents (str | None)

• contents_hash (str | None)

• is_updated (bool)

contents: str | None

The fetched contents of the product page.

contents_hash: str | None

The SHA-256 hash of the fetched contents.

is_updated: bool

Flag indicating whether the contents have been updated.

class ProductPage(location_id=None, data=None, client=None)

Bases: object

Product page object that provides methods for fetching, updating, and managing product data on the page. May be used as an asynchronous context manager.

Parameters:

• location_id (Optional[int])

• data (Optional[ProductPageData])

• client (Optional[ProductDataFetchClient])

property data: ProductPageData

Product page data model.

property context: dict[Any, Any]

Product page context data.

property client: ProductDataFetchClient

Product data fetch client.

async set_client(client)

Set the product data fetch client.

This method is asynchronous and closes the current session if there is an active session.

Parameters:

client (ProductDataFetchClient) – The new product data fetch client.

Return type:

None

subscribe_for_update(handler, event=None, call_safe=True, handler_done_callback=None)

Subscribe a handler to specific product update event(s). The handler will be invoked when the event is posted, with the event context passed as an argument.

The handler can be an asynchronous function, method, or any callable object that accepts exactly one argument (a ProductUpdateContext object) and returns None or a coroutine that resolves to None.

Parameters:

• handler (Handler) – The function or callable to invoke for the event(s).

• event (Union[ProductUpdateEvent, Iterable[ProductUpdateEvent], None], optional) – The type of product update event(s) to subscribe to. If None, the handler will be subscribed to all events.

• call_safe (bool, optional) – If True, exceptions raised by the handler are caught and logged. If False, exceptions are propagated and must be handled by the caller. Defaults to True.

• handler_done_callback (Optional[Callable[[asyncio.Future], Any]]) – Optional function to be called when the handler completes execution. Depending on the type of the handler, the callback receives an asyncio.Task or asyncio.Future object as its argument, which represents the return value of the callback execution. Defaults to None.

Raises:

TypeError – If the handler does not have a valid signature.

Return type:

None

unsubscribe_from_update(handler=None, event=None)

Unsubscribe a handler from specific product update event(s), or all handlers if no specific handler is provided. The unsubscribed handler will no longer be invoked when the event is posted.

Parameters:

• handler (Handler) – The handler to be unsubscribed from the event(s). if None, all handlers for the event are unsubscribed.

• event (Union[ProductUpdateEvent, Iterable[ProductUpdateEvent], None], optional) – The type of product update event(s) to unsubscribe from. If None, the handler(s) will be subscribed from all events.

Return type:

None

is_subscribed_for_update(handler=None, event=None)

Check if there are any subscribers for the given event(s).

Parameters:

• handler (Optional[Handler], optional) – The handler to check for subscription. If None, all handlers are checked.

• event (Union[ProductUpdateEvent, Iterable[ProductUpdateEvent], None], optional) – The type of product update event(s) to check for subscribers. If None, all events are checked.

Returns:

True if there are subscribers for the event, False otherwise.

Return type:

bool

async start_session()

Start an aiohttp client session if one is not already started.

Return type:

None

async close_session()

Close the aiohttp client session if one is open.

Return type:

None

async fetch()

Fetch the contents of the product page and extract the product data. This method does not update the internal state of the page, nor does it trigger any event handlers.

Returns:

List of product data extracted from the contents.

Return type:

list[Product]

async update_silently()

Fetch the contents of the product page, extract the product data, and update the internal state of the page without triggering any event handlers.

Return type:

None

async update(await_handlers=False, **kwargs)

Fetch the contents of the product page, extract the product data, update the internal state of the page, and trigger event handlers.

Parameters:

• await_handlers (bool, optional) – If True, the method will wait for all event handlers to complete execution. Defaults to False.

• kwargs (Any)

Return type:

None

async update_forever(interval=10.0, await_handlers=False, **kwargs)

Update the product page at regular intervals.

This method is a coroutine that runs indefinitely, updating the product page at regular intervals.

Parameters:

• interval (float, optional) – The time interval in seconds between updates. Defaults to 10.0.

• await_handlers (bool, optional) – If True, the method will wait for all event handlers to complete execution. Defaults to False.

• kwargs (Any)

Return type:

None

init_update_forever_task(interval=10.0, await_handlers=False, **kwargs)

Initialize the update forever task. If the task is already running, the method does nothing.

This method is not a coroutine. It creates a new task from the update_forever coroutine with the asyncio.create_task function. The task is stored internally and can be cancelled with the cancel_update_forever method.

Parameters:

• interval (float, optional) – The time interval in seconds between updates. Defaults to 10.0.

• await_handlers (bool, optional) – If True, the method will wait for all event handlers to complete execution. Defaults to False.

• kwargs (Any)

Returns:

The task object created by asyncio.create_task.

Return type:

asyncio.Task

async await_update_handlers()

Wait for all event handlers to complete execution.

Return type:

None

async cancel_update_handlers()

Cancel all running event handlers.

Return type:

None

async cancel_update_forever()

Cancel the update forever task if it is running.

Return type:

None

find_product(constraint=None, **attributes)

Find a product on the page that matches the specified attributes.

Attributes are specific product state information and should match the product data model fields, such as product_id, name, category, etc. A constraint function can be provided to filter products based on additional criteria or more complex conditions.

Parameters:

constraint (Optional[Callable[[Product], bool]], optional) – Optional function that takes a Product instance as input and returns a boolean indicating whether a certain constraint is met for this instance.

Returns:

The product object if found, None otherwise.

Return type:

Optional[Product]

find_products(constraint=None, **attributes)

Find products on the page that match the specified attributes.

Attributes are specific product state information and should match the product data model fields, such as product_id, name, category, etc. A constraint function can be provided to filter products based on additional criteria or more complex conditions.

Parameters:

constraint (Optional[Callable[[Product], bool]], optional) – Optional function that takes a Product instance as input and returns a boolean indicating whether a certain constraint is met for this instance.

Returns:

List of product objects that match the specified

attributes.

Return type:

list[Product]

pydantic model ProductPageData

Bases: BaseModel

Data model of a product page.

Show JSON schema

{
   "title": "ProductPageData",
   "description": "Data model of a product page.",
   "type": "object",
   "properties": {
      "locationId": {
         "title": "Locationid",
         "type": "integer"
      },
      "htmlHash": {
         "default": "",
         "title": "Htmlhash",
         "type": "string"
      },
      "products": {
         "additionalProperties": {
            "$ref": "#/$defs/Product"
         },
         "title": "Products",
         "type": "object"
      }
   },
   "$defs": {
      "Product": {
         "description": "Represents a FreshPoint.cz web page product with various attributes.\n\nArgs:\n    id_ (int):\n        Unique identifier or the product.\n    name (str):\n        Name of the product. Defaults to an empty string value.\n    category (str):\n        Category of the product. Defaults to an empty string value.\n    is_vegetarian (bool):\n        Indicates whether the product is vegetarian. Defaults to False.\n    is_gluten_free (bool):\n        Indicates whether the product is gluten-free. Defaults to False.\n    quantity (int):\n        Quantity of product items in stock. Defaults to 0.\n    price_full (float):\n        Full price of the product. If not provided, matches the current\n        selling price if the latter is provided or is set to 0 otherwise.\n    price_curr (float):\n        Current selling price. If not provided, matches the full price\n        if the latter is provided or is set to 0 otherwise.\n    pic_url (str):\n        URL of the product image. Default URL is used if not provided.\n    location_id (int):\n        Unique identifier or the product page URL. Defaults to 0.\n    location (str):\n        Name of the product location. Defaults to an empty string value.\n    timestamp (int):\n        Timestamp of the product instance initialization.\n        Defaults to the time of instantiation.",
         "properties": {
            "id": {
               "title": "Id",
               "type": "integer"
            },
            "name": {
               "default": "",
               "title": "Name",
               "type": "string"
            },
            "category": {
               "default": "",
               "title": "Category",
               "type": "string"
            },
            "isVegetarian": {
               "default": false,
               "title": "Isvegetarian",
               "type": "boolean"
            },
            "isGlutenFree": {
               "default": false,
               "title": "Isglutenfree",
               "type": "boolean"
            },
            "quantity": {
               "default": 0,
               "minimum": 0,
               "title": "Quantity",
               "type": "integer"
            },
            "priceFull": {
               "default": 0.0,
               "minimum": 0.0,
               "title": "Pricefull",
               "type": "number"
            },
            "priceCurr": {
               "default": 0.0,
               "minimum": 0.0,
               "title": "Pricecurr",
               "type": "number"
            },
            "picUrl": {
               "default": "https://images.weserv.nl/?url=http://freshpoint.freshserver.cz/backend/web/media/photo/1_f587dd3fa21b22.jpg",
               "title": "Picurl",
               "type": "string"
            },
            "locationId": {
               "default": 0,
               "title": "Locationid",
               "type": "integer"
            },
            "location": {
               "default": "",
               "title": "Location",
               "type": "string"
            },
            "timestamp": {
               "title": "Timestamp",
               "type": "number"
            }
         },
         "required": [
            "id"
         ],
         "title": "Product",
         "type": "object"
      }
   },
   "required": [
      "locationId"
   ]
}

Config:

• alias_generator: function = <function to_camel at 0x7f261f65d8a0>

• populate_by_name: bool = True

Fields:

• html_hash (str)

• location_id (int)

• products (dict[int, freshpointsync.product._product.Product])

field html_hash: str = '' (alias 'htmlHash')

SHA-256 hash of the HTML contents of the product page.

field location_id: int [Required] (alias 'locationId')

ID of the product location.

field products: dict[int, Product] [Optional]

Dictionary of products’ data on the page.

property location: str

Name of the product location. Infers from the first product in the products dictionary. If the dictionary is empty, returns an empty string.

property location_lowercase_ascii: str

Lowercase ASCII representation of the location name.

property product_categories: list[str]

List of string product categories on the page.

property product_names: list[str]

List of string product names on the page.

property url: str

URL of the product page.

class ProductPageHub(data=None, client=None, enable_multiprocessing=False)

Bases: object

Product page hub object that provides methods for managing multiple product pages at once. Each page retains its own state and can be accessed individually. Page data updates are done in parallel using asyncio tasks to optimize performance. May be used as an asynchronous context manager.

Parameters:

• data (Optional[ProductPageHubData])

• client (Optional[ProductDataFetchClient])

• enable_multiprocessing (bool)

property data: ProductPageHubData

Product page hub data model.

property client: ProductDataFetchClient

Product data fetch client.

async set_client(client)

Set the product data fetch client.

This method is asynchronous and closes the current session if there is an active session.

Parameters:

client (ProductDataFetchClient) – The new product data fetch client.

Return type:

None

subscribe_for_update(handler, event=None, call_safe=True, handler_done_callback=None)

Subscribe a handler to specific product update event(s) for all pages in the hub. The handler will be invoked when the event is posted, with the event context passed as an argument.

The handler can be an asynchronous function, method, or any callable object that accepts exactly one argument (a ProductUpdateContext object) and returns None or a coroutine that resolves to None.

Parameters:

• handler (Handler) – The function or callable to invoke for the event(s).

• event (Union[ProductUpdateEvent, Iterable[ProductUpdateEvent], None], optional) – The type of product update event(s) to subscribe to. If None, the handler will be subscribed to all events.

• call_safe (bool, optional) – If True, exceptions raised by the handler are caught and logged. If False, exceptions are propagated and must be handled by the caller. Defaults to True.

• handler_done_callback (Optional[Callable[[asyncio.Future], Any]]) – Optional function to be called when the handler completes execution. Depending on the type of the handler, the callback receives an asyncio.Task or asyncio.Future object as its argument, which represents the return value of the callback execution. Defaults to None.

Raises:

TypeError – If the handler does not have a valid signature.

Return type:

None

unsubscribe_from_update(handler=None, event=None)

Unsubscribe a handler from specific product update event(s) for all pages in the hub, or all handlers if no specific handler is provided. The unsubscribed handler will no longer be invoked when the event is posted.

Parameters:

• handler (Handler) – The handler to be unsubscribed from the event(s). if None, all handlers for the event are unsubscribed.

• event (Union[ProductUpdateEvent, Iterable[ProductUpdateEvent], None], optional) – The type of product update event(s) to unsubscribe from. If None, the handler(s) will be subscribed from all events.

Return type:

None

is_subscribed_for_update(handler=None, event=None)

Check if there are any subscribers for the given event(s) for any page in the hub.

Parameters:

• handler (Optional[Handler], optional) – The handler to check for subscription. If None, all handlers are checked.

• event (Union[ProductUpdateEvent, Iterable[ProductUpdateEvent], None], optional) – The type of product update event(s) to check for subscribers. If None, all events are checked.

Returns:

True if there are subscribers for the event, False otherwise.

Return type:

bool

set_context(key, value)

Set a context key-value pair for all pages in the hub.

Parameters:

• key (Any) – Context key.

• value (Any) – Context value.

Return type:

None

del_context(key)

Delete a context key-value pair for all pages in the hub. If the key does not exist in the page context, the method does nothing.

Parameters:

key (Any) – Context key.

Return type:

None

async start_session()

Start an aiohttp client session if one is not already started.

Return type:

None

async close_session()

Close the aiohttp client session if one is open.

Return type:

None

async new_page(location_id, fetch_contents=False, trigger_handlers=False)

Create a new product page and register it in the hub.

Parameters:

• location_id (int) – ID of the product location.

• fetch_contents (bool, optional) – If True, the page contents are fetched and updated. If False, the page contents are empty. Defaults to False.

• trigger_handlers (bool, optional) – If True, the event handlers are triggered after the page is updated. Defaults to False.

Returns:

The newly created product page object.

Return type:

ProductPage

async add_page(page, update_contents=False, trigger_handlers=False)

Add an existing product page to the hub. The page retains its own state, but receives a common client. Its contents and event handlers are updated, too.

Parameters:

• page (ProductPage) – The product page object to add.

• update_contents (bool, optional) – If True, the page contents are fetched and updated. If False, the page contents remain as is.

• trigger_handlers (bool, optional) – If True, the event handlers are triggered after the page is updated. Defaults to False.

Return type:

None

get_page(location_id)

Get a registered product page by location ID.

Parameters:

location_id (int) – ID of the product location.

Raises:

KeyError – If the page is not found.

Returns:

The product page object.

Return type:

ProductPage

get_pages()

Get all registered product pages.

Returns:

Dictionary of product page objects with

location IDs as keys.

Return type:

dict[int, ProductPage]

async remove_page(location_id, await_handlers=False)

Remove a product page from the hub.

This method unregisters the page from the hub, creates a new client for the page, and cancels (or awaits) all event handlers. It acts similarly to the dictionary pop method.

Parameters:

• location_id (int) – ID of the product location.

• await_handlers (bool, optional) – If True, the method will wait for all event handlers bound to the page to complete execution. Defaults to False.

Raises:

KeyError – If the page is not found.

Returns:

The removed product page object.

Return type:

ProductPage

async scan(start=0, stop=500, step=1)

Scan for new product pages in a range of location IDs. The pages that are valid and have products are registered in the hub.

Parameters:

• start (int, optional) – Start location ID. Defaults to 0.

• stop (int, optional) – Stop location ID. Defaults to 500.

• step (int, optional) – Step size for location IDs. Defaults to 1.

Return type:

None

async update_silently()

Fetch the contents of all product pages in the hub, extract the product data, and update the internal state of the pages without triggering any event handlers.

Return type:

None

async update(await_handlers=False, **kwargs)

Fetch the contents of all product pages in the hub, extract the product data, update the internal state of the pages, and trigger event handlers.

Parameters:

await_handlers (bool, optional) – If True, the method will wait for all event handlers to complete execution. Defaults to False.

Return type:

None

async update_forever(interval=10.0, await_handlers=False, **kwargs)

Update all product pages in the hub at regular intervals.

This method is a coroutine that runs indefinitely, updating the product page at regular intervals.

Parameters:

• interval (float, optional) – The time interval in seconds between updates. Defaults to 10.0.

• await_handlers (bool, optional) – If True, the method will wait for all event handlers to complete execution. Defaults to False.

• kwargs (Any)

Return type:

None

async init_update_forever_tasks(interval=10.0, await_handlers=False, **kwargs)

Initialize the update forever task for all product pages in the hub. If a task is already running, the method does nothing.

This method is not a coroutine. It creates a new task from the update_forever coroutine with the asyncio.create_task function. The task is stored internally and can be cancelled with the cancel_update_forever method. Note that the task is created for the hub, not for individual pages.

Parameters:

• interval (float, optional) – The time interval in seconds between updates. Defaults to 10.0.

• await_handlers (bool, optional) – If True, the method will wait for all event handlers to complete execution. Defaults to False.

• kwargs (Any)

Return type:

Task

async await_update_handlers()

Wait for all event handlers to complete execution.

Return type:

None

async cancel_update_handlers()

Cancel all running event handlers.

Return type:

None

async cancel_update_forever()

Cancel the update forever task if it is running.

Return type:

None

pydantic model ProductPageHubData

Bases: BaseModel

Data model of a product page hub.

Show JSON schema

{
   "title": "ProductPageHubData",
   "description": "Data model of a product page hub.",
   "type": "object",
   "properties": {
      "pages": {
         "additionalProperties": {
            "$ref": "#/$defs/ProductPageData"
         },
         "title": "Pages",
         "type": "object"
      }
   },
   "$defs": {
      "Product": {
         "description": "Represents a FreshPoint.cz web page product with various attributes.\n\nArgs:\n    id_ (int):\n        Unique identifier or the product.\n    name (str):\n        Name of the product. Defaults to an empty string value.\n    category (str):\n        Category of the product. Defaults to an empty string value.\n    is_vegetarian (bool):\n        Indicates whether the product is vegetarian. Defaults to False.\n    is_gluten_free (bool):\n        Indicates whether the product is gluten-free. Defaults to False.\n    quantity (int):\n        Quantity of product items in stock. Defaults to 0.\n    price_full (float):\n        Full price of the product. If not provided, matches the current\n        selling price if the latter is provided or is set to 0 otherwise.\n    price_curr (float):\n        Current selling price. If not provided, matches the full price\n        if the latter is provided or is set to 0 otherwise.\n    pic_url (str):\n        URL of the product image. Default URL is used if not provided.\n    location_id (int):\n        Unique identifier or the product page URL. Defaults to 0.\n    location (str):\n        Name of the product location. Defaults to an empty string value.\n    timestamp (int):\n        Timestamp of the product instance initialization.\n        Defaults to the time of instantiation.",
         "properties": {
            "id": {
               "title": "Id",
               "type": "integer"
            },
            "name": {
               "default": "",
               "title": "Name",
               "type": "string"
            },
            "category": {
               "default": "",
               "title": "Category",
               "type": "string"
            },
            "isVegetarian": {
               "default": false,
               "title": "Isvegetarian",
               "type": "boolean"
            },
            "isGlutenFree": {
               "default": false,
               "title": "Isglutenfree",
               "type": "boolean"
            },
            "quantity": {
               "default": 0,
               "minimum": 0,
               "title": "Quantity",
               "type": "integer"
            },
            "priceFull": {
               "default": 0.0,
               "minimum": 0.0,
               "title": "Pricefull",
               "type": "number"
            },
            "priceCurr": {
               "default": 0.0,
               "minimum": 0.0,
               "title": "Pricecurr",
               "type": "number"
            },
            "picUrl": {
               "default": "https://images.weserv.nl/?url=http://freshpoint.freshserver.cz/backend/web/media/photo/1_f587dd3fa21b22.jpg",
               "title": "Picurl",
               "type": "string"
            },
            "locationId": {
               "default": 0,
               "title": "Locationid",
               "type": "integer"
            },
            "location": {
               "default": "",
               "title": "Location",
               "type": "string"
            },
            "timestamp": {
               "title": "Timestamp",
               "type": "number"
            }
         },
         "required": [
            "id"
         ],
         "title": "Product",
         "type": "object"
      },
      "ProductPageData": {
         "description": "Data model of a product page.",
         "properties": {
            "locationId": {
               "title": "Locationid",
               "type": "integer"
            },
            "htmlHash": {
               "default": "",
               "title": "Htmlhash",
               "type": "string"
            },
            "products": {
               "additionalProperties": {
                  "$ref": "#/$defs/Product"
               },
               "title": "Products",
               "type": "object"
            }
         },
         "required": [
            "locationId"
         ],
         "title": "ProductPageData",
         "type": "object"
      }
   }
}

Config:

• alias_generator: function = <function to_camel at 0x7f261f65d8a0>

• populate_by_name: bool = True

Fields:

• pages (dict[int, freshpointsync.page._page.ProductPageData])

field pages: dict[int, ProductPageData] [Optional]

Dictionary of product page data models.

freshpointsync.parser

The parser module provides classes for parsing HTML content and extracting product data in the form of Pydantic models. It is a part of the low-level API.

class ProductPageHTMLParser(page_html)

Bases: object

A parser for processing HTML contents of a FreshPoint.cz web page.

This class uses BeautifulSoup to parse HTML contents and extract data related to the products listed on the page. The parser can search for products by either name, ID, or both.

Parameters:

page_html (str)

property page_id: int

Page ID (extracted from the page HTML <script/> tag with the “deviceId” text).

property location_name: str

The name of the location (extracted from the page HTML <title/> tag).

property products: tuple[Product, ...]

A tuple of Product instances parsed from the page HTML.

find_product(name=None, id=None)

Find a single product based on the specified name and/or ID.

Parameters:

• name (str | None) – The name of the product to filter by. Note that product names are normalized to lowercase ASCII characters for matching, allowing for partial and case-insensitive matches. If None, name filtering is not applied.

• id (int | None) – The ID of the product to filter by. The ID match is exact. If None, ID filtering is not applied.

Returns:

A Product object with the specified name and/or ID.

Return type:

Product

Raises:

ValueError – If the product with the specified name and/or ID is not found or if multiple products match the criteria (i.e., the result is not unique).

find_products(name=None)

Find a list of products based on the specified name. If the name is not specified, all products on the page are returned.

Parameters:

name (str | None) – The name of the product to filter by. Note that product names are normalized to lowercase ASCII characters for matching, allowing for partial and case-insensitive matches. If None, retrieves all products.

Returns:

Product objects with the specified name.

Return type:

tuple[Product]

class ProductFinder

Bases: object

A utility for searching and filtering products based on certain attributes and constraints. This class provides static methods to find either a single product or a list of products from an collection of Product instances.

classmethod product_matches(product, constraint=None, **attributes)

Check if a product matches the given attributes and an optional constraint.

Parameters:

• product (Product) – The product to check.

• constraint (Optional[Callable[[Product], bool]]) – An optional function that takes a Product instance as input and returns a boolean indicating whether a certain constraint is met for this instance.

• **attributes (Any) – Arbitrary keyword arguments representing the product attributes and properties and their expected values for the product to match.

Returns:

True if the product matches the given attributes and constraint, False otherwise.

Return type:

bool

classmethod find_product(products, constraint=None, **attributes)

Find a single product in an iterable of products that matches the given attributes and an optional constraint.

Parameters:

• products (Iterable[Product]) – An iterable collection of Product instances.

• constraint (Optional[Callable[[Product], bool]]) – An optional function that takes a Product instance as input and returns a boolean indicating whether a certain constraint is met for this instance.

• **attributes (Any) – Arbitrary keyword arguments representing the product attributes and properties and their expected values for the product to match.

Returns:

The first product in the iterable that matches the given attributes and constraint, or None if no such product is found.

Return type:

Optional[Product]

classmethod find_products(products, constraint=None, **attributes)

Find all products in an iterable of products that match the given attributes and an optional constraint.

Parameters:

• products (Iterable[Product]) – An iterable collection of Product instances.

• constraint (Optional[Callable[[Product], bool]]) – An optional function that takes a Product instance as input and returns a boolean indicating whether a certain constraint is met for this instance.

• **attributes (Any) – Arbitrary keyword arguments representing the product attributes and properties and their expected values for the products to match.

Returns:

A list of all products in the iterable that match the given attributes and constraint.

Return type:

list[Product]

normalize_text(text)

Normalize the given text by removing diacritics, leading/trailing whitespace, and converting it to lowercase. Non-string values are converted to strings. None values are converted to empty strings.

Parameters:

text (Any) – The text to be normalized.

Returns:

The normalized text.

Return type:

str

hash_text(text)

Calculate the SHA-256 hash of the given text.

Parameters:

text (str) – The text to be hashed.

Returns:

The SHA-256 hash of the input text in hexadecimal format.

Return type:

str

freshpointsync.product

The product module provides a Pydantic model for representing FreshPoint product data. It is a part of the high-level API. It also provides classes for storing product data change analysis results, which can be accessed from the low-level API.

class DiffValueTuple(value_self, value_other)

Bases: NamedTuple

Holds differing attribute values between two products.

Parameters:

• value_self (Any)

• value_other (Any)

value_self: Any

Value of the attribute in the first product.

value_other: Any

Value of the attribute in the second product.

pydantic model Product

Bases: BaseModel

Represents a FreshPoint.cz web page product with various attributes.

Parameters:

• id (int) – Unique identifier or the product.

• name (str) – Name of the product. Defaults to an empty string value.

• category (str) – Category of the product. Defaults to an empty string value.

• is_vegetarian (bool) – Indicates whether the product is vegetarian. Defaults to False.

• is_gluten_free (bool) – Indicates whether the product is gluten-free. Defaults to False.

• quantity (int) – Quantity of product items in stock. Defaults to 0.

• price_full (float) – Full price of the product. If not provided, matches the current selling price if the latter is provided or is set to 0 otherwise.

• price_curr (float) – Current selling price. If not provided, matches the full price if the latter is provided or is set to 0 otherwise.

• pic_url (str) – URL of the product image. Default URL is used if not provided.

• location_id (int) – Unique identifier or the product page URL. Defaults to 0.

• location (str) – Name of the product location. Defaults to an empty string value.

• timestamp (int) – Timestamp of the product instance initialization. Defaults to the time of instantiation.

Show JSON schema

{
   "title": "Product",
   "description": "Represents a FreshPoint.cz web page product with various attributes.\n\nArgs:\n    id_ (int):\n        Unique identifier or the product.\n    name (str):\n        Name of the product. Defaults to an empty string value.\n    category (str):\n        Category of the product. Defaults to an empty string value.\n    is_vegetarian (bool):\n        Indicates whether the product is vegetarian. Defaults to False.\n    is_gluten_free (bool):\n        Indicates whether the product is gluten-free. Defaults to False.\n    quantity (int):\n        Quantity of product items in stock. Defaults to 0.\n    price_full (float):\n        Full price of the product. If not provided, matches the current\n        selling price if the latter is provided or is set to 0 otherwise.\n    price_curr (float):\n        Current selling price. If not provided, matches the full price\n        if the latter is provided or is set to 0 otherwise.\n    pic_url (str):\n        URL of the product image. Default URL is used if not provided.\n    location_id (int):\n        Unique identifier or the product page URL. Defaults to 0.\n    location (str):\n        Name of the product location. Defaults to an empty string value.\n    timestamp (int):\n        Timestamp of the product instance initialization.\n        Defaults to the time of instantiation.",
   "type": "object",
   "properties": {
      "id": {
         "title": "Id",
         "type": "integer"
      },
      "name": {
         "default": "",
         "title": "Name",
         "type": "string"
      },
      "category": {
         "default": "",
         "title": "Category",
         "type": "string"
      },
      "isVegetarian": {
         "default": false,
         "title": "Isvegetarian",
         "type": "boolean"
      },
      "isGlutenFree": {
         "default": false,
         "title": "Isglutenfree",
         "type": "boolean"
      },
      "quantity": {
         "default": 0,
         "minimum": 0,
         "title": "Quantity",
         "type": "integer"
      },
      "priceFull": {
         "default": 0.0,
         "minimum": 0.0,
         "title": "Pricefull",
         "type": "number"
      },
      "priceCurr": {
         "default": 0.0,
         "minimum": 0.0,
         "title": "Pricecurr",
         "type": "number"
      },
      "picUrl": {
         "default": "https://images.weserv.nl/?url=http://freshpoint.freshserver.cz/backend/web/media/photo/1_f587dd3fa21b22.jpg",
         "title": "Picurl",
         "type": "string"
      },
      "locationId": {
         "default": 0,
         "title": "Locationid",
         "type": "integer"
      },
      "location": {
         "default": "",
         "title": "Location",
         "type": "string"
      },
      "timestamp": {
         "title": "Timestamp",
         "type": "number"
      }
   },
   "required": [
      "id"
   ]
}

Config:

• alias_generator: function = <function to_camel at 0x7f261f65d8a0>

• populate_by_name: bool = True

• validate_assignment: bool = True

Fields:

• category (str)

• id_ (int)

• is_gluten_free (bool)

• is_vegetarian (bool)

• location (str)

• location_id (int)

• name (str)

• pic_url (str)

• price_curr (float)

• price_full (float)

• quantity (int)

• timestamp (float)

field category: str = ''

Category of the product.

field id_: int [Required]

Unique identifier or the product.

field is_gluten_free: bool = False (alias 'isGlutenFree')

Indicates if the product is gluten-free.

field is_vegetarian: bool = False (alias 'isVegetarian')

Indicates if the product is vegetarian.

field location: str = ''

Name of the product location.

field location_id: int = 0 (alias 'locationId')

Unique identifier of the product page URL.

field name: str = ''

Name of the product.

field pic_url: str = 'https://images.weserv.nl/?url=http://freshpoint.freshserver.cz/backend/web/media/photo/1_f587dd3fa21b22.jpg' (alias 'picUrl')

URL of the product image.

field price_curr: NonNegativeFloat = 0.0 (alias 'priceCurr')

Current selling price of the product.

Constraints:

• ge = 0

field price_full: NonNegativeFloat = 0.0 (alias 'priceFull')

Full price of the product.

Constraints:

• ge = 0

field quantity: NonNegativeInt = 0

Quantity of product items in stock.

Constraints:

• ge = 0

field timestamp: float [Optional]

Timestamp of the product creation.

compare_price(new)

Compare the pricing details of this product instance with those of a newer instance of the same product.

This comparison is meaningful primarily when the new argument represents the same product but in a different pricing state, such as after a price adjustment.

Parameters:

new (Product) – The instance of the product to compare against. It should represent the same product at a different state or time.

Returns:

An object containing information about

changes in pricing between this product and the provided product. It includes information on changes in full price, current price, discount rates, and flags indicating the start or end of a sale.

Return type:

ProductPriceUpdateInfo

compare_quantity(new)

Compare the stock quantity of this product instance with the one of a newer instance of the same product.

This comparison is meaningful primarily when the new argument represents the same product at a different state or time, such as after a stock update.

Parameters:

new (Product) – The instance of the product to compare against. It should represent the same product at a different state or time.

Returns:

An object containing information about

changes in stock quantity of this product when compared to the provided product. It provides insights into changes in stock quantity, such as decreases, increases, depletion, or restocking.

Return type:

ProductStockUpdateInfo

diff(other, **kwargs)

Compare this product with another to identify differences, excluding the creation timestamps.

Parameters:

other (Product) – The product to compare against.

Returns:

A dictionary with keys as attribute names and

values as namedtuples containing the differing values between this product and the other product.

Return type:

dict[str, DiffValue]

is_newer_than(other)

Determine if this product is newer that the given one by comparing their creation timestamps.

Parameters:

other (Product) – The product to compare against.

Returns:

True if this product is newer than the other product,

False otherwise.

Return type:

bool

model_post_init(_Product__context)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

Return type:

None

property category_lowercase_ascii: str

Lowercase ASCII representation of the product category.

property discount_rate: float

Discount rate (<0; 1>) of the product, calculated based on the difference between the full price and the current selling price.

property is_available: bool

A product is considered available if its quantity is greater than zero.

property is_last_piece: bool

A product is considered available if its quantity equals one.

property is_on_sale: bool

A product is considered on sale if its current selling price is lower than its full price.

property is_sold_out: bool

A product is considered available if its quantity equals zero.

property location_lowercase_ascii: str

Lowercase ASCII representation of the product location name.

property name_lowercase_ascii: str

Lowercase ASCII representation of the product name.

class ProductPriceUpdateInfo(price_full_decrease=0.0, price_full_increase=0.0, price_curr_decrease=0.0, price_curr_increase=0.0, discount_rate_decrease=0.0, discount_rate_increase=0.0, sale_started=False, sale_ended=False)

Bases: object

Summarizes the details of pricing changes of a product, as determined by comparing two instances of this product.

Parameters:

• price_full_decrease (float)

• price_full_increase (float)

• price_curr_decrease (float)

• price_curr_increase (float)

• discount_rate_decrease (float)

• discount_rate_increase (float)

• sale_started (bool)

• sale_ended (bool)

price_full_decrease: float = 0.0

Decrease in the full price of the product, representing the difference between its old full price and its new full price. A value of 0.0 indicates no decrease.

price_full_increase: float = 0.0

Increase of the full price of the product, representing the difference between its new full price and its old full price. A value of 0.0 indicates no increase.

price_curr_decrease: float = 0.0

Decrease in the current selling price of the product, representing the difference between its old selling price and its new selling price. A value of 0.0 indicates no decrease.

price_curr_increase: float = 0.0

Increase in the current selling price of the product, representing the difference between its new selling price and its old selling price. A value of 0.0 indicates no increase.

discount_rate_decrease: float = 0.0

Decrease in the discount rate of the product, indicating the reduction of the discount rate in the new product compared to the old product. A value of 0.0 indicates that the discount rate has not decreased.

discount_rate_increase: float = 0.0

Increase in the discount rate of the product, indicating the increment of the discount rate in the new product compared to the old product. A value of 0.0 indicates that the discount rate has not increased.

sale_started: bool = False

A flag indicating whether a sale has started on the product. True if the new product is on sale and the old product was not.

sale_ended: bool = False

A flag indicating whether a sale has ended on the product. True if the new product is not on sale and the old product was.

class ProductQuantityUpdateInfo(stock_decrease=0, stock_increase=0, stock_depleted=False, stock_restocked=False)

Bases: object

Summarizes the details of stock quantity changes in a product, as determined by comparing two instances of this product.

Parameters:

• stock_decrease (int)

• stock_increase (int)

• stock_depleted (bool)

• stock_restocked (bool)

stock_decrease: int = 0

Decrease in stock quantity, representing how many items are fewer in the new product compared to the old product. A value of 0 implies no decrease.

stock_increase: int = 0

Increase in stock quantity, indicating how many items are more in the new product compared to the old product. A value of 0 implies no increase.

stock_depleted: bool = False

A flag indicating complete depletion of the product stock. True if the new product’s stock quantity is zero while the old product’s stock was greater than zero.

stock_restocked: bool = False

A flag indicating the product has been restocked. True if the new product’s stock quantity is greater than zero while the old product’s stock was zero.

freshpointsync.update

The update module provides classes for updating and managing changes in FreshPoint product data. It is a part of the low-level API.

class ProductUpdateContext(_ProductUpdateContext__kwargs)

Bases: Mapping

An update event context data wrapper for product update events with a mapping-like access to the data along with direct access to specific product state information via the product_old and product_new keys and the event type via the event key.

This class is designed to encapsulate the update event context data passed to event handlers during product update events.

Parameters:

_ProductUpdateContext__kwargs (dict[Any, Any])

property event: ProductUpdateEvent

The type of product update event that occurred.

property product_id: int

ID of the product being updated. If not available, defaults to 0.

property product_name: str

Name of the product being updated. If not available, defaults to an empty string.

property product_name_lowercase_ascii: str

Lowercase ASCII representation of the product name. If not available, defaults to an empty string.

property location_id: int

ID of the product location. If not available, defaults to 0.

property location_name: str

Name of the product location. If not available, defaults to an empty string.

property location_name_lowercase_ascii: str

Lowercase ASCII representation of the location name. If not available, defaults to an empty string.

property product_new: Product | None

New state of the product after the update. If not available, defaults to None.

property product_old: Product | None

Previous state of the product before the update. If not available, defaults to None.

asdict()

Convert the context data to a dictionary.

Returns:

A dictionary representation of the context data.

Return type:

dict[str, Any]

class ProductUpdateEvent(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

An enumeration of different types of product updates.

This enum is used to represent various kinds of changes or updates that can occur to a product on the product page. Each member represents a specific type of update event for easy identification and handling.

PRODUCT_ADDED = 'product_added'

Indicates that a new product has been listed on the product page.

PRODUCT_UPDATED = 'product_updated'

Indicates any update to a product’s information and state.

QUANTITY_UPDATED = 'quantity_updated'

Indicates that the stock quantity of a product has been updated.

PRICE_UPDATED = 'price_updated'

Indicates that the pricing of a product has changed, including both full price and current sale price.

OTHER_UPDATED = 'other_updated'

Indicates an update to the product’s metadata, for example, a change of the illustration picture URL.

PRODUCT_REMOVED = 'product_removed'

Indicates that a product has been removed from the product page.

class ProductUpdateEventPublisher

Bases: object

A publisher and a subscription manager for product update events.

This class maintains a registry of event handlers for different types of product update events. It allows for subscribing or unsubscribing handlers to specific events and facilitates broadcasting events to all relevant handlers.

subscribe(handler, event=None, call_safe=True, handler_done_callback=None)

Subscribe a handler to specific product update event(s). The handler will be invoked when the event is posted, with the event context passed as an argument.

The handler can be an asynchronous function, method, or any callable object that accepts exactly one argument (a ProductUpdateContext object) and returns None or a coroutine that resolves to None.

Parameters:

• handler (Handler) – The function or callable to invoke for the event(s).

• event (Union[ProductUpdateEvent, Iterable[ProductUpdateEvent], None], optional) – The type of product update event(s) to subscribe to. If None, the handler will be subscribed to all events.

• call_safe (bool, optional) – If True, exceptions raised by the handler are caught and logged. If False, exceptions are propagated and must be handled by the caller. Defaults to True.

• handler_done_callback (Optional[Callable[[asyncio.Future], Any]]) – Optional function to be called when the handler completes execution. Depending on the type of the handler, the callback receives an asyncio.Task or asyncio.Future object as its argument, which represents the return value of the callback execution. Defaults to None.

Raises:

TypeError – If the handler does not have a valid signature.

Return type:

None

unsubscribe(handler=None, event=None)

Unsubscribe a handler from specific product update event(s), or all handlers if no specific handler is provided. The unsubscribed handler will no longer be invoked when the event is posted.

Parameters:

• handler (Handler) – The handler to be unsubscribed from the event(s). if None, all handlers for the event are unsubscribed.

• event (Union[ProductUpdateEvent, Iterable[ProductUpdateEvent], None], optional) – The type of product update event(s) to unsubscribe from. If None, the handler(s) will be subscribed from all events.

Return type:

None

is_subscribed(handler=None, event=None)

Check if there are any subscribers for the given event(s).

Parameters:

• handler (Optional[Handler], optional) – The handler to check for subscription. If None, all handlers are checked.

• event (Union[ProductUpdateEvent, Iterable[ProductUpdateEvent], None], optional) – The type of product update event(s) to check for subscribers. If None, all events are checked.

Returns:

True if there are subscribers for the event, False otherwise.

Return type:

bool

get_context(event, product_new, product_old, **kwargs)

Construct a ProductUpdateContext instance from the provided product and additional parameters, merging the global context data with the specific product information and any additional keyword arguments (the latter takes precedence).

Parameters:

• product_new (Optional[Product]) – The updated product instance, or None if not applicable.

• product_old (Optional[Product]) – The original product instance before the update, or None if not applicable.

• **kwargs (Any) – Additional context parameters to be included.

• event (ProductUpdateEvent)

Returns:

A context object populated with the global context data and the provided product information and

parameters.

Return type:

ProductUpdateContext

post(event, product_new, product_old, **kwargs)

Notify registered handlers about a specific product update event.

This method constructs a ProductUpdateContext with new and old product details, along with any additional keyword arguments. It then broadcasts the event to all relevant handlers, asynchronous and synchronous alike, registered for the event type.

Parameters:

• event (ProductUpdateEvent) – The type of product update event to be broadcasted.

• product_new (Optional[Product]) – The updated product instance, or None if there’s no new product.

• product_old (Optional[Product]) – The original product instance before update, or None if unavailable.

• **kwargs (Any) – Additional keyword arguments to be included in the event context. If any of the keyword arguments overlaps with the context data, it takes precedence.

Return type:

None

class ProductCacheUpdater(products, publisher)

Bases: object

Product cache manager. Updates the cache and notifies subscribers about product updates.

This class is responsible for maintaining a cache of product data and using an event publisher to notify interested parties of changes to product information. It provides methods to create, delete, and update products in the cache and to trigger appropriate events for each action.

Parameters:

• products (dict[int, Product])

• publisher (ProductUpdateEventPublisher)

create_product(product, **kwargs)

Add a new product to the cache and post a PRODUCT_ADDED event.

Note that this method assumes that the product is not present in the cache already.

Parameters:

• product (Product) – the product to add to the cache.

• **kwargs (Any) – Additional keyword arguments to include in the update event context.

Return type:

None

delete_product(product, **kwargs)

Remove a product from the cache and posts a PRODUCT_REMOVED event.

Note that this method assumes that the product is present in the cache.

Parameters:

• product (Product) – the product to remove from the cache.

• **kwargs (Any) – Additional keyword arguments to include in the update event context.

Return type:

None

update_product(product, product_cached, diff, **kwargs)

Update a product in the cache and post events reflecting the specific changes occured.

This method determines what aspects of the product have changed (e.g., stock level, price, etc.) and posts the appropriate events for each type of update.

Note that this method assumes that the product is different from the cached product.

Parameters:

• product (Product) – The new product state after the update.

• product_cached (Product) – The original product state before the update.

• **kwargs (Any) – Additional keyword arguments to include in the update event context.

• diff (dict[str, DiffValueTuple])

• **kwargs

Return type:

None

async update(products, await_handlers=False, **kwargs)

Processe a batch of product updates, adding, deleting, or updating products as necessary.

This method iterates through a given iterable of products, updating the cache to reflect new, removed, or changed products. It optionally waits for all event handlers to complete if await_handlers is True.

Parameters:

• products (Iterable[Product]) – products to process.

• await_handlers (bool) – If True, waits for all triggered event handlers to complete.

• **kwargs (Any) – Additional keyword arguments to include in the update event context.

Return type:

None

update_silently(products)

Update the product cache without triggering any events.

This method clears the cache dictionary and subsequently populates it with the given products, effectively replacing the entire cache contents while maintaining the dictionary object itself.

Parameters:

products (Iterable[Product]) – products to populate the cache with.

Return type:

None

class HandlerValidator

Bases: object

A utility class for validating handler functions or methods based on their signature and whether they are asynchronous or synchronous.

classmethod is_valid_async_handler(handler)

Check if a handler is an asynchronous coroutine function with a valid signature (accepts exactly one argument).

Parameters:

handler (Any) – The handler to be validated.

Returns:

True if the handler is valid and asynchronous, otherwise, False.

Return type:

bool

classmethod is_valid_sync_handler(handler)

Check if a handler is a synchronous function or method with a valid signature (accepts exactly one argument).

Parameters:

handler (Any) – The handler to be validated.

Returns:

True if the handler is valid and synchronous, otherwise, False.

Return type:

bool

classmethod is_valid_handler(handler)

Check if a handler has a valid signature and is a valid function or method, regardless of whether it is synchronous or asynchronous.

Parameters:

handler (Any) – The handler to be validated.

Returns:

True if the handler is valid, otherwise, False.

Return type:

bool

is_valid_handler(handler)

Validate a handler based on its signature and whether it is an asynchronous coroutine or a synchronous function or method.

Parameters:

handler (Any) – The handler to be validated.

Returns:

True if the handler is valid, otherwise, False.

Return type:

bool