data_product

data_product

ProductGenerator module-attribute

ProductGenerator = Callable[[Path], ProductList]

Callable method type. Users must provide a ProductGenerator to map over raw data.

Parameters:

Name	Type	Description	Default
path	Path	Workdir holding raw data (Should be one per run from a batch)	required

Returns:

Type	Description
ProductList	List of data products to be sorted and used to produce assets

ProductList module-attribute

ProductList = List[SerializeAsAny[InstanceOf[DataProduct]]]

List of serializable DataProduct or child classes thereof

DataProduct pydantic-model

Bases: BaseModel

Base class for data products to be generated and handled.

Attributes:

Name	Type	Description
product_type	str	Product type should be the same as the class name. The product type is used to search for products from a DataProductCollection.
tags	Tags	Tags to be used for sorting data.
metadata	dict[str, str]	A dictionary of metadata to be used as a tool tip for mousover in grafana

Show JSON schema:

{
  "additionalProperties": true,
  "description": "Base class for data products to be generated and handled.\n\nAttributes:\n    product_type (str): Product type should be the same as the class name.\n        The product type is used to search for products from a [DataProductCollection][trendify.api.DataProductCollection].\n    tags (Tags): Tags to be used for sorting data.\n    metadata (dict[str, str]): A dictionary of metadata to be used as a tool tip for mousover in grafana",
  "properties": {
    "tags": {
      "items": {
        "anyOf": []
      },
      "title": "Tags",
      "type": "array"
    },
    "metadata": {
      "additionalProperties": {
        "type": "string"
      },
      "default": {},
      "title": "Metadata",
      "type": "object"
    }
  },
  "required": [
    "tags"
  ],
  "title": "DataProduct",
  "type": "object"
}

Config:

• extra: 'allow'

Fields:

• tags (Tags)
• metadata (dict[str, str])
• product_type (str)

Validators:

• _remove_computed_fields

product_type pydantic-field

product_type: str

Returns:

Type	Description
str	Product type should be the same as the class name. The product type is used to search for products from a DataProductCollection.

__init_subclass__

__init_subclass__(**kwargs: Any) -> None

Registers child subclasses to be able to parse them from JSON file using the deserialize_child_classes method

Source code in src/trendify/api/base/data_product.py

def __init_subclass__(cls, **kwargs: Any) -> None:
    """
    Registers child subclasses to be able to parse them from JSON file using the
    [deserialize_child_classes][trendify.api.DataProduct.deserialize_child_classes] method
    """
    super().__init_subclass__(**kwargs)
    _data_product_subclass_registry[cls.__name__] = cls

append_to_list

append_to_list(l: List)

Appends self to list.

Parameters:

Name	Type	Description	Default
l	List	list to which self will be appended	required

Returns:

Type	Description
Self	returns instance of self

Source code in src/trendify/api/base/data_product.py

def append_to_list(self, l: List):
    """
    Appends self to list.

    Args:
        l (List): list to which `self` will be appended

    Returns:
        (Self): returns instance of `self`
    """
    l.append(self)
    return self

deserialize_child_classes classmethod

deserialize_child_classes(key: str, **kwargs)

Loads json data to pydandic dataclass of whatever DataProduct child time is appropriate

Parameters:

Name	Type	Description	Default
key	str	json key	required
kwargs	dict	json entries stored under given key	{}

Source code in src/trendify/api/base/data_product.py

@classmethod
def deserialize_child_classes(cls, key: str, **kwargs):
    """
    Loads json data to pydandic dataclass of whatever DataProduct child time is appropriate

    Args:
        key (str): json key
        kwargs (dict): json entries stored under given key
    """
    type_key = "product_type"
    elements = kwargs.get(key, None)
    if elements:
        for index in range(len(kwargs[key])):
            duck_info = kwargs[key][index]
            if isinstance(duck_info, dict):
                product_type = duck_info.pop(type_key)
                duck_type = _data_product_subclass_registry[product_type]
                kwargs[key][index] = duck_type(**duck_info)