Msgspec benchmark

benchmark/factories.py

@@ -9,7 +9,7 @@ from benchmark.pydantic_benchmark.models import (
 )
 
 
-COLUMNS_COUNT = 100
+COLUMNS_COUNT = 1000
 PLANOGRAMS_COUNT = 1000
 
 

benchmark/msgspec_benchmark/benchmark.py

@@ -0,0 +1,13 @@
+import msgspec
+from benchmark.base import BenchmarkBase
+from benchmark.msgspec_benchmark.models import PlanogramsBulkInputPayload
+
+
+class MsgSpecBenchmark(BenchmarkBase):
+    def _benchmark(self) -> None:
+        test_data = self._read_test_file()
+
+        # decoder = msgspec.json.Decoder(PlanogramsBulkInputPayload).decode
+        # decoder.loads(test_data)
+
+        data = msgspec.json.decode(test_data, type=PlanogramsBulkInputPayload)

benchmark/msgspec_benchmark/models.py

@@ -0,0 +1,51 @@
+from enum import Enum
+from typing import Annotated, Optional
+from uuid import uuid4
+from msgspec import Meta, Struct, field
+
+
+QuantityInt = Annotated[int, Meta(ge=0, le=2147483647)]
+StrictSmallInt = Annotated[int, Meta(ge=0, le=32767)]
+PriceFloat = Annotated[float, Meta(ge=0, le=99999999.99)]
+ExternalId = Annotated[str, Meta(min_length=1, max_length=32)]
+
+
+class ColumnItemType(str, Enum):
+    PRODUCT = "PRODUCT"
+    COMPONENT = "COMPONENT"
+
+
+class CorrelationId(Struct, rename="camel"):
+    correlation_id: str = field(default_factory=lambda: uuid4().hex)
+
+
+class ColumnsInput(Struct, rename="camel"):
+    column_number: StrictSmallInt
+    external_product_id: Optional[ExternalId] = field(default=None)
+    old_qty: Optional[QuantityInt] = field(default_factory=lambda: None)
+    new_qty: Optional[QuantityInt] = field(default_factory=lambda: None)
+    old_price: Optional[PriceFloat] = field(default_factory=lambda: None)
+    new_price: Optional[PriceFloat] = field(default_factory=lambda: None)
+    select_map: Optional[list[StrictSmallInt]] = field(default_factory=lambda: None)
+    item_type: Optional[ColumnItemType] = field(
+        default_factory=lambda: ColumnItemType.PRODUCT
+    )
+
+
+class PlanogramInput(CorrelationId, Struct, rename="camel"):
+    machine_external_id: ExternalId = field(default="")
+    columns: list[ColumnsInput] = field(default_factory=list)
+
+    # class Config:
+    #     title = "Planogram"
+    #     alias_generator = to_camel_case
+    #     populate_by_name = True
+    #     str_strip_whitespace = True
+
+
+class PlanogramsBulkInputPayload(Struct, rename="camel"):
+    planograms: list[PlanogramInput] = field(default_factory=list)
+
+    # class Config:
+    #     populate_by_name = True
+    #     alias_generator = to_camel_case