MPT-14767 E2E seeding proof of concept

pyproject.toml

@@ -26,6 +26,7 @@ dependencies = [
 
 [dependency-groups]
 dev = [
+    "dependency-injector>=4.48.2",
     "freezegun==1.5.*",
     "ipdb==0.13.*",
     "ipython==9.*",

seed/catalog/catalog.py

@@ -0,0 +1,39 @@
+import asyncio
+import logging
+from typing import Any
+
+from seed.catalog.item import seed_items
+from seed.catalog.item_group import seed_item_group
+from seed.catalog.product import seed_product
+from seed.catalog.product_parameters import seed_parameters
+from seed.catalog.product_parameters_group import seed_parameter_group
+
+logger = logging.getLogger(__name__)
+
+
+async def seed_groups_and_group_params() -> None:
+    """Seed parallel tasks for item groups and parameter groups."""
+    tasks: list[asyncio.Task[Any]] = [
+        asyncio.create_task(seed_item_group()),
+        asyncio.create_task(seed_parameter_group()),
+    ]
+    await asyncio.gather(*tasks)
+
+
+async def seed_items_and_params() -> None:
+    """Seed final tasks for items and parameters."""
+    tasks: list[asyncio.Task[Any]] = [
+        asyncio.create_task(seed_items()),
+        asyncio.create_task(seed_parameters()),
+    ]
+    await asyncio.gather(*tasks)
+
+
+async def seed_catalog() -> None:
+    """Seed catalog data including products, item groups, and parameters."""
+    logger.debug("Seeding catalog ...")
+    await seed_product()
+    await seed_groups_and_group_params()
+    await seed_items_and_params()
+
+    logger.debug("Seeded catalog completed.")

seed/catalog/item.py

@@ -0,0 +1,135 @@
+import logging
+from typing import Any
+
+from dependency_injector.wiring import inject
+
+from mpt_api_client import AsyncMPTClient
+from mpt_api_client.resources.catalog.items import Item
+from seed.context import Context
+from seed.defaults import (
+    DEFAULT_CONTEXT,
+    DEFAULT_MPT_OPERATIONS,
+    DEFAULT_MPT_VENDOR,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@inject
+async def refresh_item(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_vendor: AsyncMPTClient = DEFAULT_MPT_VENDOR,
+) -> Item | None:
+    """Refresh item in context (always fetch)."""
+    item_id = context.get_string("catalog.item.id")
+    if not item_id:
+        return None
+    item_resource = await mpt_vendor.catalog.items.get(item_id)
+    context["catalog.item.id"] = item_resource.id
+    context.set_resource("catalog.item", item_resource)
+    return item_resource
+
+
+@inject
+async def get_item(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_vendor: AsyncMPTClient = DEFAULT_MPT_VENDOR,
+) -> Item | None:
+    """Get item from context or fetch from API if not cached."""
+    item_id = context.get_string("catalog.item.id")
+    if not item_id:
+        return None
+    try:
+        catalog_item = context.get_resource("catalog.item", item_id)
+    except ValueError:
+        catalog_item = None
+    if not isinstance(catalog_item, Item):
+        logger.debug("Loading item: %s", item_id)
+        catalog_item = await mpt_vendor.catalog.items.get(item_id)
+        context["catalog.item.id"] = catalog_item.id
+        context.set_resource("catalog.item", catalog_item)
+        return catalog_item
+    return catalog_item
+
+
+@inject
+def build_item(context: Context = DEFAULT_CONTEXT) -> dict[str, Any]:
+    """Build item data dictionary for creation."""
+    product_id = context.get("catalog.product.id")
+    item_group_id = context.get("catalog.item_group.id")
+    return {
+        "product": {"id": product_id},
+        "parameters": [],
+        "name": "Product Item 1",
+        "description": "Product Item 1 - Description",
+        "group": {"id": item_group_id},
+        "unit": {
+            "id": "UNT-1229",
+            "name": "<string> 1",
+            "revision": 1,
+            "description": "<string>TEST",
+            "statistics": {"itemCount": 34},
+        },
+        "terms": {"model": "quantity", "period": "1m", "commitment": "1m"},
+        "quantityNotApplicable": False,
+        "externalIds": {"vendor": "item_1"},
+    }
+
+
+@inject
+async def create_item(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_vendor: AsyncMPTClient = DEFAULT_MPT_VENDOR,
+) -> Item:
+    """Create item and cache in context."""
+    item_data = build_item(context=context)
+    catalog_item = await mpt_vendor.catalog.items.create(item_data)
+    context["catalog.item.id"] = catalog_item.id
+    context.set_resource("catalog.item", catalog_item)
+    return catalog_item
+
+
+@inject
+async def review_item(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_vendor: AsyncMPTClient = DEFAULT_MPT_VENDOR,
+) -> Item | None:
+    """Review item if in draft status and cache result."""
+    logger.debug("Reviewing catalog.item ...")
+    catalog_item = context.get_resource("catalog.item")
+    if catalog_item.status != "Draft":
+        return catalog_item  # type: ignore[return-value]
+    catalog_item = await mpt_vendor.catalog.items.review(catalog_item.id)
+    context.set_resource("catalog.item", catalog_item)
+    return catalog_item
+
+
+@inject
+async def publish_item(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+) -> Item | None:
+    """Publish item if in reviewing status and cache result."""
+    logger.debug("Publishing catalog.item ...")
+    catalog_item = context.get_resource("catalog.item")
+    if catalog_item.status != "Reviewing":
+        return catalog_item  # type: ignore[return-value]
+    catalog_item = await mpt_operations.catalog.items.publish(catalog_item.id)
+    context.set_resource("catalog.item", catalog_item)
+    return catalog_item
+
+
+@inject
+async def seed_items(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_vendor: AsyncMPTClient = DEFAULT_MPT_VENDOR,
+    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+) -> None:
+    """Seed catalog items (create/review/publish)."""
+    logger.debug("Seeding catalog.item ...")
+    existing = await refresh_item(context=context, mpt_vendor=mpt_vendor)
+    if not existing:
+        await create_item(context=context, mpt_vendor=mpt_vendor)
+    await review_item(context=context, mpt_vendor=mpt_vendor)
+    await publish_item(context=context, mpt_operations=mpt_operations)
+    logger.debug("Seeded catalog.item completed.")

seed/catalog/item_group.py

@@ -0,0 +1,87 @@
+import logging
+from typing import Any
+
+from dependency_injector.wiring import inject
+
+from mpt_api_client import AsyncMPTClient
+from mpt_api_client.resources.catalog.products_item_groups import ItemGroup
+from seed.context import Context
+from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_VENDOR
+
+logger = logging.getLogger(__name__)
+
+
+@inject
+async def get_item_group(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_vendor: AsyncMPTClient = DEFAULT_MPT_VENDOR,
+) -> ItemGroup | None:
+    """Get item group from context or fetch from API."""
+    item_group_id = context.get_string("catalog.item_group.id")
+    if not item_group_id:
+        return None
+    try:
+        item_group = context.get_resource("catalog.item_group", item_group_id)
+    except ValueError:
+        item_group = None
+    if not isinstance(item_group, ItemGroup):
+        logger.debug("Refreshing item group: %s", item_group_id)
+        product_id = context.get_string("catalog.product.id")
+        item_group = await mpt_vendor.catalog.products.item_groups(product_id).get(item_group_id)
+        return set_item_group(item_group, context=context)
+    return item_group
+
+
+@inject
+def set_item_group(
+    item_group: ItemGroup,
+    context: Context = DEFAULT_CONTEXT,
+) -> ItemGroup:
+    """Set item group in context."""
+    context["catalog.item_group.id"] = item_group.id
+    context.set_resource("catalog.item_group", item_group)
+    return item_group
+
+
+@inject
+def build_item_group(context: Context = DEFAULT_CONTEXT) -> dict[str, Any]:
+    """Build item group data dictionary."""
+    product_id = context.get("catalog.product.id")
+    return {
+        "product": {"id": product_id},
+        "name": "Items",
+        "label": "Items",
+        "description": "Default item group",
+        "displayOrder": 100,
+        "default": True,
+        "multiple": True,
+        "required": True,
+    }
+
+
+@inject
+async def init_item_group(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_vendor: AsyncMPTClient = DEFAULT_MPT_VENDOR,
+) -> ItemGroup:
+    """Get or create item group."""
+    item_group = await get_item_group()
+
+    if not item_group:
+        logger.debug("Creating item group ...")
+        product_id = context.get_string("catalog.product.id")
+        item_group_data = build_item_group()
+        item_group = await mpt_vendor.catalog.products.item_groups(product_id).create(
+            item_group_data
+        )
+        logger.debug("Item group created: %s", item_group.id)
+        return set_item_group(item_group)
+    logger.debug("Item group found: %s", item_group.id)
+    return item_group
+
+
+async def seed_item_group() -> None:
+    """Seed item group."""
+    logger.debug("Seeding catalog.item_group ...")
+    await init_item_group()
+    logger.debug("Seeded catalog.item_group completed.")

seed/catalog/product.py

@@ -0,0 +1,95 @@
+import logging
+import pathlib
+
+from dependency_injector.wiring import inject
+
+from mpt_api_client import AsyncMPTClient
+from mpt_api_client.resources.catalog.products import Product
+from seed.context import Context
+from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_OPERATIONS, DEFAULT_MPT_VENDOR
+
+icon = pathlib.Path(__file__).parent / "FIL-9920-4780-9379.png"
+
+logger = logging.getLogger(__name__)
+
+namespace = "catalog.product"
+
+
+@inject
+async def get_product(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_vendor: AsyncMPTClient = DEFAULT_MPT_VENDOR,
+) -> Product | None:
+    """Get product from context or fetch from API."""
+    product_id = context.get_string(f"{namespace}.id")
+    if not product_id:
+        return None
+    try:
+        product = context.get_resource(namespace, product_id)
+    except ValueError:
+        product = None
+    if not isinstance(product, Product):
+        logger.debug("Refreshing product: %s", product_id)
+        product = await mpt_vendor.catalog.products.get(product_id)
+        context.set_resource(namespace, product)
+        context[f"{namespace}.id"] = product.id
+        return product
+    return product
+
+
+@inject
+async def init_product(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_vendor: AsyncMPTClient = DEFAULT_MPT_VENDOR,
+) -> Product:
+    """Get or create product."""
+    product = await get_product()
+    if not product:
+        logger.debug("Creating product ...")
+        with pathlib.Path.open(icon, "rb") as icon_file:
+            product = await mpt_vendor.catalog.products.create(
+                {"name": "Test Product", "website": "https://www.example.com"}, icon=icon_file
+            )
+            context.set_resource(namespace, product)
+            context[f"{namespace}.id"] = product.id
+        logger.debug("Product created: %s", product.id)
+    return product
+
+
+@inject
+async def review_product(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_vendor: AsyncMPTClient = DEFAULT_MPT_VENDOR,
+) -> Product | None:
+    """Review product if in draft status."""
+    product = await get_product()
+    if not product or product.status != "Draft":
+        return product
+    logger.debug("Reviewing product: %s", product.id)
+    product = await mpt_vendor.catalog.products.review(product.id)
+    context.set_resource(namespace, product)
+    return product
+
+
+@inject
+async def publish_product(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+) -> Product | None:
+    """Publish product if in reviewing status."""
+    product = await get_product()
+    if not product or product.status != "Reviewing":
+        return product
+    logger.debug("Publishing product: %s", product.id)
+    product = await mpt_operations.catalog.products.publish(product.id)
+    context.set_resource(namespace, product)
+    return product
+
+
+async def seed_product() -> None:
+    """Seed product data."""
+    logger.debug("Seeding catalog.product ...")
+    await init_product()
+    await review_product()
+    await publish_product()
+    logger.debug("Seeded catalog.product completed.")