fix: Fix typing in examples and tests (#166)

Make mypy checking more strict for Pydantic models and fix or suppress the resulting errors.

Author: spanglerco

examples/assetmanagement/assets.py

@@ -1,3 +1,5 @@
+from datetime import datetime, timezone
+
 from nisystemlink.clients.assetmanagement import AssetManagementClient
 from nisystemlink.clients.assetmanagement.models import (
     AssetBusType,
@@ -27,7 +29,7 @@
         model_name="NI PXIe-6368",
         serial_number="01BB877A",
         vendor_name="NI",
-        vendor_number="4244",
+        vendor_number=4244,
         bus_type=AssetBusType.ACCESSORY,
         name="PCISlot2",
         asset_type=AssetType.DEVICE_UNDER_TEST,
@@ -41,7 +43,7 @@
         self_calibration=SelfCalibration(
             temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
             is_limited=False,
-            date="2022-06-07T18:58:05.000Z",
+            date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
         ),
         is_NI_asset=True,
         workspace="your-workspace-id",
@@ -50,19 +52,23 @@
         ),
         external_calibration=ExternalCalibration(
             temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
-            date="2022-06-07T18:58:05.000Z",
+            date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
             recommended_interval=10,
-            next_recommended_date="2023-11-14T20:42:11.583Z",
-            next_custom_due_date="2024-11-14T20:42:11.583Z",
-            resolved_due_date="2022-06-07T18:58:05.000Z",
+            next_recommended_date=datetime(
+                2023, 11, 14, 20, 42, 11, 583000, tzinfo=timezone.utc
+            ),
+            next_custom_due_date=datetime(
+                2024, 11, 14, 20, 42, 11, 583000, tzinfo=timezone.utc
+            ),
+            resolved_due_date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
         ),
         properties={"Key1": "Value1"},
         keywords=["Keyword1"],
         discovery_type=AssetDiscoveryType.MANUAL,
         file_ids=["608a5684800e325b48837c2a"],
         supports_self_test=True,
         supports_reset=True,
-        partNumber="A1234 B5",
+        part_number="A1234 B5",
     )
 ]
 
@@ -80,8 +86,8 @@
     take=1,
     descending=False,
     calibratable_only=False,
-    returnCount=False,
-)
+    return_count=False,
+)  # type: ignore
 client.query_assets(query=query_asset_request)
 
 # Link files to the created asset.

examples/dataframe/create_write_data.py

@@ -31,13 +31,15 @@
 
 # Append via explicit AppendTableDataRequest (JSON)
 frame_request = DataFrame(
-    data=[[i, random.random(), datetime.now().isoformat()] for i in range(3)]
+    data=[[str(i), str(random.random()), datetime.now().isoformat()] for i in range(3)]
 )
 client.append_table_data(table_id, AppendTableDataRequest(frame=frame_request))
 
 # Append via DataFrame model directly (JSON)
 frame_direct = DataFrame(
-    data=[[i + 3, random.random(), datetime.now().isoformat()] for i in range(3)]
+    data=[
+        [str(i + 3), str(random.random()), datetime.now().isoformat()] for i in range(3)
+    ]
 )
 client.append_table_data(table_id, frame_direct)
 

examples/dataframe/export_data.py

@@ -4,6 +4,8 @@
 from nisystemlink.clients.dataframe.models import (
     ColumnFilter,
     ColumnOrderBy,
+    ColumnType,
+    DataType,
     ExportFormat,
     ExportTableDataRequest,
     FilterOperation,
@@ -14,13 +16,26 @@
 # List a table
 response = client.list_tables(take=1)
 table = response.tables[0]
+index_column = next(
+    (col for col in table.columns if col.column_type == ColumnType.Index)
+)
+order_column = next(
+    (col for col in table.columns if col.name != index_column.name), index_column
+)
+filter_value = (
+    "0001-01-01T00:00:00Z" if index_column.data_type == DataType.Timestamp else "0"
+)
 
 # Export table data with query options
 request = ExportTableDataRequest(
-    columns=["col1"],
-    order_by=[ColumnOrderBy(column="col2", descending=True)],
+    columns=[index_column.name],
+    order_by=[ColumnOrderBy(column=order_column.name, descending=True)],
     filters=[
-        ColumnFilter(column="col1", operation=FilterOperation.NotEquals, value="0")
+        ColumnFilter(
+            column=index_column.name,
+            operation=FilterOperation.NotEquals,
+            value=filter_value,
+        )
     ],
     response_format=ExportFormat.CSV,
 )

examples/dataframe/query_read_data.py

@@ -1,5 +1,6 @@
 from nisystemlink.clients.dataframe import DataFrameClient
 from nisystemlink.clients.dataframe.models import (
+    ColumnType,
     DecimationMethod,
     DecimationOptions,
     QueryDecimatedDataRequest,
@@ -10,17 +11,22 @@
 # List a table
 response = client.list_tables(take=1)
 table = response.tables[0]
+y_column = next(
+    (col for col in table.columns if col.column_type != ColumnType.Index),
+    table.columns[0],
+)
 
 # Get table metadata by table id
 client.get_table_metadata(table.id)
 
 # Query decimated table data
 request = QueryDecimatedDataRequest(
     decimation=DecimationOptions(
-        x_column="index",
-        y_columns=["col1"],
+        x_column=None,
+        y_columns=[y_column.name],
         intervals=1,
         method=DecimationMethod.MaxMin,
     )
 )
-client.query_decimated_data(table.id, request)
+rows = client.query_decimated_data(table.id, request)
+print(rows.frame.model_dump())

examples/spec/get_and_query_specs.py

@@ -5,9 +5,9 @@
     ConditionRange,
     ConditionType,
     CreateSpecificationsRequest,
+    CreateSpecificationsRequestObject,
     NumericConditionValue,
     QuerySpecificationsRequest,
-    SpecificationDefinition,
     SpecificationLimit,
     SpecificationType,
 )
@@ -22,7 +22,7 @@
 # Create the spec requests
 product = "Amplifier"
 spec_requests = [
-    SpecificationDefinition(
+    CreateSpecificationsRequestObject(
         product_id=product,
         spec_id="spec1",
         type=SpecificationType.PARAMETRIC,
@@ -31,7 +31,7 @@
         limit=SpecificationLimit(min=1.2, max=1.5),
         unit="mV",
     ),
-    SpecificationDefinition(
+    CreateSpecificationsRequestObject(
         product_id=product,
         spec_id="spec2",
         type=SpecificationType.PARAMETRIC,
@@ -58,7 +58,7 @@
             ),
         ],
     ),
-    SpecificationDefinition(
+    CreateSpecificationsRequestObject(
         product_id=product,
         spec_id="spec3",
         type=SpecificationType.FUNCTIONAL,

examples/spec/update_and_delete_specs.py

@@ -1,10 +1,12 @@
+from typing import cast
+
 from nisystemlink.clients.core import HttpConfiguration
 from nisystemlink.clients.spec import SpecClient
 from nisystemlink.clients.spec.models import (
     QuerySpecificationsRequest,
-    SpecificationDefinition,
     SpecificationType,
     UpdateSpecificationsRequest,
+    UpdateSpecificationsRequestObject,
 )
 
 # Setup the server configuration to point to your instance of SystemLink Enterprise
@@ -27,22 +29,22 @@
     print(f"Original spec1 block: {original_spec1.block}")
     print(f"Original spec1 version: {original_spec1.version}")
 
-# make the modifications
-modified_spec = SpecificationDefinition(
-    id=original_spec1.id,
-    product_id=original_spec1.product_id,
-    spec_id=original_spec1.spec_id,
-    type=SpecificationType.FUNCTIONAL,
-    keywords=["work", "reviewed"],
-    block="modifiedBlock",
-    version=original_spec1.version,
-    workspace=original_spec1.workspace,
-)
-update_response = client.update_specs(
-    specs=UpdateSpecificationsRequest(specs=[modified_spec])
-)
-if update_response and update_response.updated_specs:
-    print(f"New spec1 version: {update_response.updated_specs[0].version}")
+    # make the modifications
+    modified_spec = UpdateSpecificationsRequestObject(
+        id=cast(str, original_spec1.id),
+        product_id=cast(str, original_spec1.product_id),
+        spec_id=cast(str, original_spec1.spec_id),
+        type=SpecificationType.FUNCTIONAL,
+        keywords=["work", "reviewed"],
+        block="modifiedBlock",
+        version=cast(int, original_spec1.version),
+        workspace=cast(str, original_spec1.workspace),
+    )
+    update_response = client.update_specs(
+        specs=UpdateSpecificationsRequest(specs=[modified_spec])
+    )
+    if update_response and update_response.updated_specs:
+        print(f"New spec1 version: {update_response.updated_specs[0].version}")
 
 # query again to see new version
 response = client.query_specs(

examples/test_plan/test_plan_templates.py

@@ -29,7 +29,7 @@
         test_program="TP-INT-002",
         estimated_duration_in_seconds=86400,
         system_filter="os:linux AND arch:x64",
-        dut_Filter="modelName = 'cRIO-9045' AND serialNumber = '01E82ED0'",
+        dut_filter="modelName = 'cRIO-9045' AND serialNumber = '01E82ED0'",
         execution_actions=[
             ManualExecution(action="boot", type="MANUAL"),
             JobExecution(

examples/test_plan/test_plans.py

@@ -35,7 +35,7 @@
         dut_serial_number="serial_number_123",
         test_program="TP-Integration-001",
         system_filter="os:linux AND arch:x64",
-        dut_Filter="modelName = 'cRIO-9045' AND serialNumber = '01E82ED0'",
+        dut_filter="modelName = 'cRIO-9045' AND serialNumber = '01E82ED0'",
         workspace="your_workspace_id",
         file_ids_from_template=["file1", "file2"],
         dashboard=Dashboard(
@@ -69,7 +69,7 @@
 
 # Query test plan using id.
 query_test_plans_request = QueryTestPlansRequest(
-    skip=0, take=1, descending=False, returnCount=False
+    take=1, descending=False, return_count=False
 )
 client.query_test_plans(query_request=query_test_plans_request)
 

examples/testmonitor/steps.py

@@ -1,3 +1,5 @@
+from typing import cast
+
 from nisystemlink.clients.testmonitor import TestMonitorClient
 from nisystemlink.clients.testmonitor.models import (
     CreateResultRequest,
@@ -44,7 +46,7 @@ def create_test_result():
 create_response = create_test_result()
 
 # Create the step requests
-result_id = create_response.results[0].result_id
+result_id = cast(str, create_response.results[0].id)
 step_requests = [
     CreateStepRequest(
         step_id="step1",
@@ -89,7 +91,7 @@ def create_test_result():
     steps=[
         UpdateStepRequest(
             step_id=step.step_id,
-            result_id=step.result_id,
+            result_id=cast(str, step.result_id),
             data=StepData(
                 text="My output string",
                 parameters=[
@@ -116,7 +118,9 @@ def create_test_result():
 # delete all steps at once
 delete_response = client.delete_steps(
     steps=[
-        StepIdResultIdPair(step_id=step.step_id, result_id=step.result_id)
+        StepIdResultIdPair(
+            step_id=cast(str, step.step_id), result_id=cast(str, step.result_id)
+        )
         for step in queried_steps
     ]
 )

mypy.ini

@@ -5,6 +5,11 @@ warn_unused_configs=True
 plugins = pydantic.mypy
 ignore_missing_imports=True
 
+[pydantic-mypy]
+init_typed=True
+warn_required_dynamic_aliases=True
+warn_untyped_fields=True
+
 [mypy-nisystemlink.*]
 disallow_untyped_calls=True
 disallow_untyped_defs=True