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 or parsing webpage HTML content.

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

freshpointsync.Product
freshpointsync.ProductPage
freshpointsync.ProductPageData
freshpointsync.ProductPageHub
freshpointsync.ProductPageHubData
freshpointsync.ProductUpdateEvent
freshpointsync.is_valid_handler()
freshpointsync.logger

`freshpointsync.client`

freshpointsync.client package provides means for fetching HTML content of Freshpoint webpages. It is a part of the low-level API.

class ProductDataFetchClient[source]

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.

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

The base URL of the FreshPoint.cz website.

classmethod get_page_url(location_id)[source]

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 timeout: ClientTimeout

Client request timeout.

set_timeout(timeout)[source]

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.ClientTimeout 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)[source]

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

property session: ClientSession | None

The aiohttp.ClientSession 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 start_session(force=False)[source]

Start a new aiohttp client session and create an SSL context.

If a session is already started, a new session is not created unless the force parameter is set to True. If the SSL context is already created, it is not recreated.

Parameters:

force (bool, optional) – If True, forcefully close the existing session and start a new one. If no session is started, this parameter has no effect. Defaults to False.

Return type:

None

async set_session(session)[source]

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

async close_session()[source]

Close the aiohttp client session if one is open.

If the session is already closed, this method has no effect on the client session object. If an SSL context has been created, it is also cleared after closing the session.

Return type:

None

async fetch(location_id, timeout=None, max_retries=None)[source]

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`

freshpoint.page package provides means for interacting with FreshPoint webpages on the product level.

Classes from this package are key components of the high-level API and are are also available in the top-level freshpointsync package for easier access.

class FetchInfo(contents, contents_hash, is_updated)[source]

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)[source]

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 (int | None)

data (ProductPageData | None)

client (ProductDataFetchClient | None)

async start_session()[source]

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

Return type:

None

async close_session()[source]

Close the aiohttp client session if one is open.

Return type:

None

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)[source]

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, call_blocking=True, handler_done_callback=None)[source]

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.

call_blocking (bool, optional) – If True, the synchronous handler is executed in a blocking manner, i.e., called directly without using an executor. If False, the synchronous handler is executed in a non-blocking manner in a separate thread. Asynchronous handlers are always executed in a non-blocking manner, this parameter has no effect on them. 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)[source]

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 subscribe to. If None, the handler will be subscribed to all events.

Return type:

None

is_subscribed_for_update(handler=None, event=None)[source]

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 subscribe to. If None, the handler will be subscribed to all events.

Returns:

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

Return type:

bool

async fetch()[source]

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(silent=False, await_handlers=False, **kwargs)[source]

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

Parameters:

silent (bool, optional) – If True, the product data is updated without triggering any event handlers. Defaults to False.

await_handlers (bool, optional) – If True, all event handlers are awaited to complete execution. This parameter has no effect if silent is True. Defaults to False.

**kwargs (Any) – Additional keyword arguments to pass to the event handlers. If the silent parameter is True, these arguments are ignored.

Return type:

None

async update_forever(interval=10.0, await_handlers=False, silent=False, **kwargs)[source]

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.

silent (bool, optional) – If True, the product data is updated without triggering any event handlers. Defaults to False.

await_handlers (bool, optional) – If True, all event handlers are awaited to complete execution. This parameter has no effect if silent is True. Defaults to False.

**kwargs (Any) – Additional keyword arguments to pass to the event handlers. If the silent parameter is True, these arguments are ignored.

Return type:

None

init_update_forever_task(interval=10.0, silent=False, await_handlers=False, **kwargs)[source]

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.

silent (bool, optional) – If True, the product data is updated without triggering any event handlers. Defaults to False.

await_handlers (bool, optional) – If True, all event handlers are awaited to complete execution. This parameter has no effect if silent is True. Defaults to False.

**kwargs (Any) – Additional keyword arguments to pass to the event handlers. If the silent parameter is True, these arguments are ignored.

Returns:

The task object created by asyncio.create_task.

Return type:

asyncio.Task

async cancel_update_forever_task()[source]

Cancel the update forever task if it is running.

Return type:

None

async await_update_handlers()[source]

Wait for all event handlers to complete execution.

Return type:

None

async cancel_update_handlers()[source]

Cancel all running event handlers.

Return type:

None

find_product(constraint=None, **attributes)[source]

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.

**attributes (Any) – Product attributes to match.

Returns:

The product object if found, None otherwise.

Return type:

Optional[Product]

find_products(constraint=None, **attributes)[source]

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.

**attributes (Any) – Product attributes to match.

Returns:

List of product objects that match the specified
attributes.

Return type:

list[Product]
pydantic model ProductPageData[source]
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"
         },
         "default": {},
         "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    info (str):\n        Additional information about the product. Defaults to an empty\n        string value.\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"
            },
            "info": {
               "default": "",
               "title": "Info",
               "type": "string"
            },
            "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 0x7f21dbd332e0>

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] = {}

Dictionary of products’ IDs and data models 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)[source]

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 (ProductPageHubData | None)

client (ProductDataFetchClient | None)

enable_multiprocessing (bool)

async start_session()[source]

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

Return type:

None

async close_session()[source]

Close the aiohttp client session if one is open.

Return type:

None

property data: ProductPageHubData

Product page hub data model.

property client: ProductDataFetchClient

Product data fetch client.

async set_client(client)[source]

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, call_blocking=True, handler_done_callback=None)[source]

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.

call_blocking (bool, optional) – If True, the synchronous handler is executed in a blocking manner, i.e., called directly without using an executor. If False, the synchronous handler is executed in a non-blocking manner in a separate thread. Asynchronous handlers are always executed in a non-blocking manner, this parameter has no effect on them. 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)[source]

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 subscribe to. If None, the handler will be subscribed to all events.

Return type:

None

is_subscribed_for_update(handler=None, event=None)[source]

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 subscribe to. If None, the handler will be subscribed to all events.

Returns:

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

Return type:

bool

set_context(key, value)[source]

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)[source]

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

property pages: Dict[int, ProductPage]

Dictionary of product page objects with location IDs as keys.

async new_page(location_id, fetch_contents=False, trigger_handlers=False)[source]

Create a new product page and register it in the hub. The page receives a common client. Its contents can be fetched and updated optionally.

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 contents are fetched. This parameter has no effect if fetch_contents is False. Defaults to False.

Returns:

The newly created product page object.

Return type:

ProductPage

async add_page(page, update_contents=False, trigger_handlers=False)[source]

Add an existing product page to the hub. The page retains its own state, but receives a common client. Its contents can be fetched and updated optionally.

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 contents are updated. This parameter has no effect if update_contents is False. Defaults to False.

Return type:

None

async remove_page(location_id, await_handlers=False)[source]

Remove a product page from the hub. The page retains its own state, but receives a new client. All event handlers are cancelled or awaited.

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. Otherwise, the method will cancel all event handlers. Defaults to False.

Raises:

KeyError – If the page is not found.

Returns:

The removed product page object.

Return type:

ProductPage

async scan(start=1, stop=999, step=1)[source]

Scan for new product pages in a range of location IDs. The pages that are valid and have products are registered in the hub. The existing pages are updated. The update handlers are not triggered.

Note: unlike Python’s range function, the stop parameter is inclusive.

Parameters:

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

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

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

Return type:

None

async update(silent=False, await_handlers=False, **kwargs)[source]

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:

silent (bool, optional) – If True, the product data is updated without triggering any event handlers. Defaults to False.

await_handlers (bool, optional) – If True, all event handlers are awaited to complete execution. This parameter has no effect if silent is True. Defaults to False.

**kwargs (Any) – Additional keyword arguments to pass to the event handlers. If the silent parameter is True, these arguments are ignored.

Return type:

None

async update_forever(interval=10.0, silent=False, await_handlers=False, **kwargs)[source]

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 the updates. Defaults to 10 seconds.

silent (bool, optional) – If True, the product data is updated without triggering any event handlers. Defaults to False.

await_handlers (bool, optional) – If True, all event handlers are awaited to complete execution. This parameter has no effect if silent is True. Defaults to False.

**kwargs (Any) – Additional keyword arguments to pass to the event handlers. If the silent parameter is True, these arguments are ignored.

Return type:

None

init_update_forever_task(interval=10.0, silent=False, await_handlers=False, **kwargs)[source]

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 the updates. Defaults to 10.0 seconds.

silent (bool, optional) – If True, the product data is updated without triggering any event handlers. Defaults to False.

await_handlers (bool, optional) – If True, all event handlers are awaited to complete execution. This parameter has no effect if silent is True. Defaults to False.

**kwargs (Any) – Additional keyword arguments to pass to the event handlers. If the silent parameter is True, these arguments are ignored.

Return type:

Task

async cancel_update_forever_task()[source]

Cancel the update forever task of the hub. Note that the separate update forever tasks of individual pages are not cancelled.

Return type:

None

async await_update_handlers()[source]

Wait for all event handlers to complete execution.

Return type:

None

async cancel_update_handlers()[source]

Cancel all running event handlers.

Return type:

None
pydantic model ProductPageHubData[source]
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"
         },
         "default": {},
         "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    info (str):\n        Additional information about the product. Defaults to an empty\n        string value.\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"
            },
            "info": {
               "default": "",
               "title": "Info",
               "type": "string"
            },
            "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"
               },
               "default": {},
               "title": "Products",
               "type": "object"
            }
         },
         "required": [
            "locationId"
         ],
         "title": "ProductPageData",
         "type": "object"
      }
   }
}
Config:

alias_generator: function = <function to_camel at 0x7f21dbd332e0>

populate_by_name: bool = True

Fields:

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

field pages: Dict[int, ProductPageData] = {}

Dictionary of product page IDs and data models.

`freshpointsync.parser`

freshpointsync.parser package provides means for parsing HTML contents of FreshPoint webpages and extracting product data. It is a part of the low-level API.

class ProductFinder[source]

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)[source]

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)[source]

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)[source]

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]

class ProductPageHTMLParser(page_html)[source]

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)[source]

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.

id_ (int | None)

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)[source]

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]

hash_text(text)[source]

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

normalize_text(text)[source]

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

parse_page_contents(page_html)[source]

Parse the HTML contents of a FreshPoint.cz web page and extract product information.

Parameters:

page_html (str) – HTML contents of the product page.

Returns:

A tuple of Product instances parsed from the page HTML.

Return type:

tuple[Product]

`freshpointsync.product`

freshpoint.product package provides a product model for representing FreshPoint product data. It also provides classes for storing product data change analysis results, which are are part of the low-level API.

The Product model is a part of the high-level API and can be accessed from the top-level freshpointsync package.

class DiffValueTuple(value_self, value_other)[source]

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[source]
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.

info (str) – Additional information about the product. Defaults to an empty string value.

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    info (str):\n        Additional information about the product. Defaults to an empty\n        string value.\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"
      },
      "info": {
         "default": "",
         "title": "Info",
         "type": "string"
      },
      "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 0x7f21dbd332e0>

populate_by_name: bool = True

validate_assignment: bool = True

Fields:

category (str)

id_ (int)

info (str)

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 info: str = ''

Additional information about 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: Annotated[float, Ge(ge=0)] = 0.0 (alias 'priceCurr')

Current selling price of the product.

Constraints:

ge = 0

field price_full: Annotated[float, Ge(ge=0)] = 0.0 (alias 'priceFull')

Full price of the product.

Constraints:

ge = 0

field quantity: Annotated[int, Ge(ge=0)] = 0

Quantity of product items in stock.

Constraints:

ge = 0

field timestamp: float [Optional]

Timestamp of the product creation.

compare_price(new)[source]

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)[source]

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)[source]

Compare this product with another to identify differences.

This method compares the fields of this product with the fields of another product instance to identify differences between them. model_dump method is used to extract the data from the product instances.

Parameters:

other (Product) – The product to compare against.

**kwargs – Additional keyword arguments to pass to the model_dump method calls of the product instances.

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)[source]

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)[source]

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.

Parameters:

_Product__context (object)

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)[source]

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)[source]

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.runner`

freshpointsync.runner package provides means for running and managing syncronous and asynchronous tasks in a non-blocking manner. It is a part of the low-level API.

class CallableRunner(executor=None)[source]

Bases: object

A utility for running asynchronous and synchronous callables in a non-blocking or blocking manner (the latter is only relevant for synchronous callables) with optional error handling and the ability to await or cancel all running tasks.

Parameters:

executor (Executor | None)

tasks: set[Task]

A set that stores all running or pending tasks associated with calls to the run_async method.

futures: set[Future]

A set that stores all running or pending futures associated with calls to the run_sync method.

executor

An optional concurrent.futures.Executor object to be used for running synchronous functions in the run_sync method.

run_async(func, *func_args, run_safe=True, done_callback=None)[source]

Schedule a function that returns a coroutine to be run, optionally with error handling and a completion callback.

This method is specifically designed for running coroutine functions that are asynchronous in nature. Providing a synchronous function to this method will fail at runtime.

Parameters:

func (Callable[..., Coroutine[Any, Any, T]]) – The coroutine function to be run.

*func_args (Any) – The arguments to run the coroutine function with.

run_safe (bool) – If True, the potential exceptions raised by the coroutine are caught and logged, and the result is set to None in case of an error. If False, exceptions are propagated and must be handled by the caller. Defaults to True.

done_callback (Optional[Callable[[asyncio.Task], Any]]) – An optional callback to be called when the task completes.

Returns:

An asyncio task object representing
the scheduled coroutine. The task can be awaited to obtain the result of the coroutine function call or cancelled.

Return type:

asyncio.Task[Optional[T]]

run_sync(func, *func_args, run_safe=True, run_blocking=True, done_callback=None)[source]

Schedule a synchronous function to be run in a blocking or a non-blocking manner, optionally with error handling and a completion callback.

This method is specifically designed for synchronous functions that block. If run_blocking is set to True, the function is executed directly without using an executor. If run_blocking is set to False, the function is executed in a non-blocking manner using an executor, allowing for concurrent execution of multiple functions. Providing an asynchronous function will fail at runtime.

Parameters:

func (Callable[..., T]) – The synchronous function to be run.

*func_args (Any) – The arguments to run the function with.

run_safe (bool) – If True, the potential exceptions raised by the synchronous function are caught and logged, and the result is set to None in case of an error. If False, exceptions are propagated and must be handled by the caller. Defaults to True.

run_blocking (bool) – If True, the synchronous function is executed in a blocking manner, i.e., called directly without using an executor. If False, the function is executed in a non-blocking manner in a separate thread. Defaults to True.

done_callback (Optional[Callable[[asyncio.Future], Any]]) – An optional callback to be called when the future completes.

Returns:

An asyncio future object representing
the scheduled execution of the synchronous function. The future can be awaited to obtain the result of the function call. Note that cancellation of the future is not possible if the function is already running (for more information, see the concurrent.futures documentation on cancellation of the future objects).

Return type:

asyncio.Future[Optional[T]]

async await_all()[source]

Wait for all scheduled asynchronous and synchronous tasks to complete.

This method gathers all active asyncio tasks and futures and awaits for their completion. It is particularly useful for ensuring that all background operations have finished before proceeding to another stage of the application or gracefully shutting down the application.

Note that this method ensures that the tracking sets of tasks and futures have been cleared after their completion, effectively resetting the runner’s state.

Return type:

None

async cancel_all()[source]

Attempt to cancel all active tasks and futures.

This method gathers and cancells all active asyncio tasks and attempts to cancel all active futures created by running synchronous functions. For the latter, the cancellation if only possible if the future has not started running yet (for more information, see the concurrent.futures documentation on cancellation of the future objects). The method is particularly useful for cancelling all background operations before proceeding to another stage of the application or gracefully shutting down the application.

Note that this method ensures that the tracking set of tasks have been cleared after their completion. The set of futures is not cleared, as the cancellation of the futures is not guaranteed to be successful.

Return type:

None

`freshpointsync.update`

frespointsync.update package provides means for updating and managing changes in FreshPoint product data. It is a part of the low-level API.

class HandlerValidator[source]

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)[source]

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

Parameters:

handler (object) – 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)[source]

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

Parameters:

handler (object) – The handler to be validated.

Returns:

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

Return type:

bool

classmethod is_valid_handler(handler)[source]

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 (object) – The handler to be validated.

Returns:

True if the handler is valid, otherwise, False.

Return type:

bool

class ProductCacheUpdater(products, publisher)[source]

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)

static get_location_id(products)[source]

Get the location ID from the given products.

Parameters:

products (Iterable[Product]) – A collection of products.

Returns:

The location ID if it is found in the products,
otherwise, None.

Return type:

Optional[int]

create_product(product, **kwargs)[source]

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)[source]

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)[source]

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.

diff (dict[str, DiffValueTuple]) – A dictionary of differences between the new and old product states.

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

Return type:

None

async update(products, await_handlers=False, **kwargs)[source]

Process 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)[source]

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 ProductUpdateContext(_ProductUpdateContext__kwargs)[source]

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[str | Any, Any])

property event: ProductUpdateEvent

The type of product update event that occurred.

property timestamp: float

The timestamp of the product update event.

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: str

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

property location_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()[source]

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)[source]

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[source]

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, run_safe=True, run_blocking=True, handler_done_callback=None)[source]

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.

run_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.

run_blocking (bool, optional) – If True, synchronous handlers are executed in a blocking manner, i.e., called directly without using an executor. If False, synchronous handlers are executed in a non-blocking manner in a separate thread. 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)[source]

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)[source]

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)[source]

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:

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

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.

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)[source]

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

is_valid_handler(handler)[source]

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

API Reference

freshpointsync.client

freshpointsync.page

freshpointsync.parser

freshpointsync.product

freshpointsync.runner