MPT-16255 Seeding for catalog

mpt_api_client/resources/catalog/products.py

@@ -65,9 +65,9 @@ class ProductsServiceConfig:
 
 
 class ProductsService(
-    CreateFileMixin[Model],
-    UpdateFileMixin[Model],
-    PublishableMixin[Model],
+    CreateFileMixin[Product],
+    UpdateFileMixin[Product],
+    PublishableMixin[Product],
     GetMixin[Product],
     DeleteMixin,
     CollectionMixin[Product],
@@ -128,9 +128,9 @@ def update_settings(self, product_id: str, settings: ResourceData) -> Product:
 
 
 class AsyncProductsService(
-    AsyncCreateFileMixin[Model],
-    AsyncUpdateFileMixin[Model],
-    AsyncPublishableMixin[Model],
+    AsyncCreateFileMixin[Product],
+    AsyncUpdateFileMixin[Product],
+    AsyncPublishableMixin[Product],
     AsyncGetMixin[Product],
     AsyncDeleteMixin,
     AsyncCollectionMixin[Product],

pyproject.toml

@@ -131,8 +131,11 @@ per-file-ignores = [
   "tests/e2e/accounts/*.py: WPS430 WPS202",
   "tests/e2e/catalog/*.py: WPS202 WPS421",
   "tests/e2e/catalog/items/*.py: WPS110 WPS202",
+  "tests/seed/catalog/test_product.py: WPS202 WPS204 WPS219",
   "tests/*: WPS432 WPS202",
-  "seed/accounts/*.py: WPS453",
+  "seed/*: WPS404",
+  "seed/accounts/*.py: WPS204 WPS404 WPS453",
+  "seed/catalog/product.py: WPS202 WPS204 WPS217 WPS201 WPS213 WPS404"
 ]
 
 [tool.ruff]

seed/accounts/api_tokens.py

@@ -1,20 +1,20 @@
 import logging
 import os
 
-from dependency_injector.wiring import inject
+from dependency_injector.wiring import Provide, inject
 
 from mpt_api_client import AsyncMPTClient
 from mpt_api_client.resources.accounts.api_tokens import ApiToken
+from seed.container import Container
 from seed.context import Context
-from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_OPERATIONS
 
 logger = logging.getLogger(__name__)
 
 
 @inject
 async def get_api_token(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_ops: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+    context: Context = Provide[Container.context],
+    mpt_ops: AsyncMPTClient = Provide[Container.mpt_operations],
 ) -> ApiToken | None:
     """Get API token from context or fetch from API."""
     api_token_id = context.get_string("accounts.api_token.id")
@@ -34,7 +34,7 @@ async def get_api_token(
 
 @inject
 def build_api_token_data(
-    context: Context = DEFAULT_CONTEXT,
+    context: Context = Provide[Container.context],
 ) -> dict[str, object]:
     """Get API token data dictionary for creation."""
     account_id = os.getenv("CLIENT_ACCOUNT_ID")
@@ -50,8 +50,8 @@ def build_api_token_data(
 
 @inject
 async def init_api_token(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_ops: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+    context: Context = Provide[Container.context],
+    mpt_ops: AsyncMPTClient = Provide[Container.mpt_operations],
 ) -> ApiToken:
     """Get or create API token."""
     api_token = await get_api_token(context=context, mpt_ops=mpt_ops)

seed/accounts/buyer.py

@@ -1,23 +1,21 @@
 import logging
 import os
-import pathlib
 
-from dependency_injector.wiring import inject
+from dependency_injector.wiring import Provide, inject
 
 from mpt_api_client import AsyncMPTClient
 from mpt_api_client.resources.accounts.buyers import Buyer
+from seed.container import Container
 from seed.context import Context
-from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_OPERATIONS
+from seed.static.static import ICON
 
 logger = logging.getLogger(__name__)
 
-icon = pathlib.Path("seed/data/logo.png").resolve()
-
 
 @inject
 async def get_buyer(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+    context: Context = Provide[Container.context],
+    mpt_operations: AsyncMPTClient = Provide[Container.mpt_operations],
 ) -> Buyer | None:
     """Get buyer from context or fetch from API."""
     buyer_id = context.get_string("accounts.buyer.id")
@@ -36,7 +34,7 @@ async def get_buyer(
 
 
 @inject
-def build_buyer_data(context: Context = DEFAULT_CONTEXT) -> dict[str, object]:
+def build_buyer_data(context: Context = Provide[Container.context]) -> dict[str, object]:
     """Build buyer data dictionary for creation."""
     buyer_account_id = os.getenv("CLIENT_ACCOUNT_ID")
     if not buyer_account_id:
@@ -65,16 +63,16 @@ def build_buyer_data(context: Context = DEFAULT_CONTEXT) -> dict[str, object]:
 
 @inject
 async def init_buyer(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+    context: Context = Provide[Container.context],
+    mpt_operations: AsyncMPTClient = Provide[Container.mpt_operations],
 ) -> Buyer:
     """Get or create buyer."""
     buyer = await get_buyer(context=context, mpt_operations=mpt_operations)
     if buyer is None:
         buyer_data = build_buyer_data(context=context)
         logger.debug("Creating buyer ...")
-        with open(str(icon), "rb") as icon_file:  # noqa: PTH123
-            created = await mpt_operations.accounts.buyers.create(buyer_data, file=icon_file)
+        with ICON.open("rb") as icon_fd:
+            created = await mpt_operations.accounts.buyers.create(buyer_data, file=icon_fd)
         if isinstance(created, Buyer):
             context.set_resource("accounts.buyer", created)
             context["accounts.buyer.id"] = created.id

seed/accounts/licensee.py

@@ -1,23 +1,21 @@
 import logging
 import os
-import pathlib
 
-from dependency_injector.wiring import inject
+from dependency_injector.wiring import Provide, inject
 
 from mpt_api_client import AsyncMPTClient
 from mpt_api_client.resources.accounts.licensees import Licensee
+from seed.container import Container
 from seed.context import Context
-from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_CLIENT
+from seed.static.static import ICON
 
 logger = logging.getLogger(__name__)
 
-icon = pathlib.Path("seed/data/logo.png").resolve()
-
 
 @inject
 async def get_licensee(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_client: AsyncMPTClient = DEFAULT_MPT_CLIENT,
+    context: Context = Provide[Container.context],
+    mpt_client: AsyncMPTClient = Provide[Container.mpt_client],
 ) -> Licensee | None:
     """Get licensee from context or fetch from API."""
     licensee_id = context.get_string("accounts.licensee.id")
@@ -37,7 +35,7 @@ async def get_licensee(
 
 @inject
 def build_licensee_data(  # noqa: WPS238
-    context: Context = DEFAULT_CONTEXT,
+    context: Context = Provide[Container.context],
 ) -> dict[str, object]:
     """Get licensee data dictionary for creation."""
     account_id = os.getenv("CLIENT_ACCOUNT_ID")
@@ -76,15 +74,15 @@ def build_licensee_data(  # noqa: WPS238
 
 @inject
 async def init_licensee(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_client: AsyncMPTClient = DEFAULT_MPT_CLIENT,
+    context: Context = Provide[Container.context],
+    mpt_client: AsyncMPTClient = Provide[Container.mpt_client],
 ) -> Licensee:
     """Get or create licensee."""
     licensee = await get_licensee(context=context, mpt_client=mpt_client)
     if licensee is None:
         licensee_data = build_licensee_data(context=context)
         logger.debug("Creating licensee ...")
-        with open(str(icon), "rb") as icon_file:  # noqa: PTH123
+        with ICON.open("rb") as icon_file:
             created = await mpt_client.accounts.licensees.create(licensee_data, file=icon_file)
         if isinstance(created, Licensee):
             context.set_resource("accounts.licensee", created)

seed/accounts/module.py

@@ -1,20 +1,20 @@
 import logging
 
-from dependency_injector.wiring import inject
+from dependency_injector.wiring import Provide, inject
 
 from mpt_api_client import AsyncMPTClient
 from mpt_api_client.resources.accounts.modules import Module
 from mpt_api_client.rql.query_builder import RQLQuery
+from seed.container import Container
 from seed.context import Context
-from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_OPERATIONS
 
 logger = logging.getLogger(__name__)
 
 
 @inject
 async def get_module(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+    context: Context = Provide[Container.context],
+    mpt_operations: AsyncMPTClient = Provide[Container.mpt_operations],
 ) -> Module | None:
     """Get module from context or fetch from API."""
     module_id = context.get_string("accounts.module.id")
@@ -34,8 +34,8 @@ async def get_module(
 
 @inject
 async def refresh_module(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+    context: Context = Provide[Container.context],
+    mpt_operations: AsyncMPTClient = Provide[Container.mpt_operations],
 ) -> Module | None:
     """Refresh module in context (always fetch)."""
     module = await get_module(context=context, mpt_operations=mpt_operations)

seed/accounts/seller.py

@@ -1,21 +1,20 @@
-# mypy: disable-error-code=unreachable
 import logging
 import uuid
 
-from dependency_injector.wiring import inject
+from dependency_injector.wiring import Provide, inject
 
 from mpt_api_client import AsyncMPTClient
 from mpt_api_client.resources.accounts.sellers import Seller
+from seed.container import Container
 from seed.context import Context
-from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_OPERATIONS
 
 logger = logging.getLogger(__name__)
 
 
 @inject
 async def get_seller(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+    context: Context = Provide[Container.context],
+    mpt_operations: AsyncMPTClient = Provide[Container.mpt_operations],
 ) -> Seller | None:
     """Get seller from context or fetch from API."""
     seller_id = context.get_string("accounts.seller.id")
@@ -33,7 +32,6 @@ async def get_seller(
     return seller
 
 
-@inject
 def build_seller_data(external_id: str | None = None) -> dict[str, object]:
     """Get seller data dictionary for creation."""
     if external_id is None:
@@ -54,11 +52,11 @@ def build_seller_data(external_id: str | None = None) -> dict[str, object]:
 
 @inject
 async def init_seller(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+    context: Context = Provide[Container.context],
+    mpt_operations: AsyncMPTClient = Provide[Container.mpt_operations],
 ) -> Seller | None:
     """Get or create seller. Returns Seller if successful, None otherwise."""
-    seller = await get_seller(context=context, mpt_operations=mpt_operations)
+    seller = await get_seller()
     if seller is None:
         logger.debug("Creating seller ...")
         seller_data = build_seller_data()
@@ -68,13 +66,12 @@ async def init_seller(
             context["accounts.seller.id"] = created.id
             logger.info("Seller created: %s", created.id)
             return created
-        logger.warning("Seller creation failed")
+        logger.warning("Seller creation failed")  # type: ignore[unreachable]
         return None
     logger.info("Seller already exists: %s", seller.id)
     return seller
 
 
-@inject
 async def seed_seller() -> None:
     """Seed seller."""
     logger.debug("Seeding seller ...")

seed/accounts/user_group.py

@@ -2,20 +2,20 @@
 import logging
 import os
 
-from dependency_injector.wiring import inject
+from dependency_injector.wiring import Provide, inject
 
 from mpt_api_client import AsyncMPTClient
 from mpt_api_client.resources.accounts.user_groups import UserGroup
+from seed.container import Container
 from seed.context import Context
-from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_OPERATIONS
 
 logger = logging.getLogger(__name__)
 
 
 @inject
 async def get_user_group(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+    context: Context = Provide[Container.context],
+    mpt_operations: AsyncMPTClient = Provide[Container.mpt_operations],
 ) -> UserGroup | None:
     """Get user group from context or fetch from API."""
     user_group_id = context.get_string("accounts.user_group.id")
@@ -35,7 +35,7 @@ async def get_user_group(
 
 @inject
 def build_user_group_data(
-    context: Context = DEFAULT_CONTEXT,
+    context: Context = Provide[Container.context],
 ) -> dict[str, object]:
     """Get user group data dictionary for creation."""
     account_id = os.getenv("CLIENT_ACCOUNT_ID")
@@ -54,8 +54,8 @@ def build_user_group_data(
 
 @inject
 async def init_user_group(
-    context: Context = DEFAULT_CONTEXT,
-    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+    context: Context = Provide[Container.context],
+    mpt_operations: AsyncMPTClient = Provide[Container.mpt_operations],
 ) -> UserGroup | None:
     """Get or create user group."""
     user_group = await get_user_group(context=context, mpt_operations=mpt_operations)

seed/catalog/authorization.py

@@ -0,0 +1,39 @@
+import uuid
+
+from dependency_injector.wiring import Provide, inject
+
+from mpt_api_client import AsyncMPTClient
+from mpt_api_client.resources.catalog.authorizations import Authorization
+from seed.container import Container
+from seed.context import Context
+from seed.helper import init_resource, require_context_id
+
+
+async def seed_authorization() -> None:
+    """Seed authorization."""
+    await init_resource("catalog.authorization.id", create_authorization)
+
+
+@inject
+async def create_authorization(
+    operations: AsyncMPTClient = Provide[Container.mpt_operations],
+    context: Context = Provide[Container.context],
+) -> Authorization:
+    """Creates an authorization."""
+    product_id = require_context_id(context, "catalog.product.id", "Create authorization")
+    seller_id = require_context_id(context, "accounts.seller.id", "Create authorization")
+    account_id = require_context_id(context, "accounts.account.id", "Create authorization")
+    short_uuid = uuid.uuid4().hex[:8]
+
+    authorization_data = {
+        "externalIds": {"operations": f"e2e-seeded-{short_uuid}"},
+        "product": {"id": product_id},
+        "owner": {"id": seller_id},
+        "journal": {"firstInvoiceDate": "2025-12-01", "frequency": "1m"},
+        "eligibility": {"client": True, "partner": True},
+        "currency": "USD",
+        "notes": "E2E Seeded",
+        "name": "E2E Seeded",
+        "vendor": {"id": account_id},
+    }
+    return await operations.catalog.authorizations.create(authorization_data)