Feat: Add MyPy support

.pre-commit-config.yaml

@@ -11,7 +11,17 @@ repos:
         args: [ --fix ]
       # Run the formatter.
       - id: ruff-format
-
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: 'v1.15.0'
+    hooks:
+      - id: mypy
+        additional_dependencies:
+            [
+                types-mock,
+                types-requests,
+                types-PyYAML,
+                pydantic
+            ]
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.4.0
     hooks:

csfunctions/actions/dummy.py

@@ -8,7 +8,4 @@ class DummyAction(BaseAction):
     Dummy Action, for unit testing
     """
 
-    def __init__(self, id: str, **kwargs):  # pylint: disable=redefined-builtin
-        super().__init__(name=ActionNames.DUMMY, id=id, data=kwargs["data"])
-
-    name: Literal[ActionNames.DUMMY]
+    name: Literal[ActionNames.DUMMY] = ActionNames.DUMMY

csfunctions/events/dummy.py

@@ -17,8 +17,5 @@ class DummyEvent(BaseEvent):
     Dummy Event, for unit testing
     """
 
-    def __init__(self, event_id: str, data: DummyEventData, **_):
-        super().__init__(name=EventNames.DUMMY, event_id=event_id, data=data)
-
-    name: Literal[EventNames.DUMMY]
-    data: DummyEventData = Field([])
+    name: Literal[EventNames.DUMMY] = EventNames.DUMMY
+    data: DummyEventData = Field(..., description="Dummy Event Data")

csfunctions/events/workflow_task_trigger.py

@@ -18,8 +18,5 @@ class WorkflowTaskTriggerEventData(BaseModel):
 
 
 class WorkflowTaskTriggerEvent(BaseEvent):
-    def __init__(self, event_id: str, data: WorkflowTaskTriggerEventData, **_):
-        super().__init__(name=EventNames.WORKFLOW_TASK_TRIGGER, event_id=event_id, data=data)
-
-    name: Literal[EventNames.WORKFLOW_TASK_TRIGGER]
+    name: Literal[EventNames.WORKFLOW_TASK_TRIGGER] = EventNames.WORKFLOW_TASK_TRIGGER
     data: WorkflowTaskTriggerEventData

csfunctions/handler.py

@@ -6,11 +6,11 @@
 from typing import Callable
 
 import yaml
-from pydantic import BaseModel
 
 from csfunctions import ErrorResponse, Event, Request, WorkloadResponse
 from csfunctions.actions import ActionUnion
 from csfunctions.config import ConfigModel, FunctionModel
+from csfunctions.events import EventData
 from csfunctions.objects import BaseObject
 from csfunctions.response import ResponseUnion
 from csfunctions.service import Service
@@ -31,7 +31,7 @@ def _get_function(function_name: str, function_dir: str) -> FunctionModel:
     config = _load_config(function_dir)
     func = next(func for func in config.functions if func.name == function_name)
     if not func:
-        raise ValueError(f"Could not find function with name { function_name} in the environment.yaml.")
+        raise ValueError(f"Could not find function with name {function_name} in the environment.yaml.")
     return func
 
 
@@ -53,7 +53,7 @@ def link_objects(event: Event):
     e.g. document.part
     """
     data = getattr(event, "data", None)
-    if not isinstance(data, BaseModel):
+    if data is None or not isinstance(data, EventData):  # type: ignore  # MyPy doesn't like PEP604
         return
 
     # we expect all objects to be passed in Event.data
@@ -81,24 +81,26 @@ def execute(function_name: str, request_body: str, function_dir: str = "src") ->
     try:
         request = Request(**json.loads(request_body))
         link_objects(request.event)
+
         function_callback = get_function_callable(function_name, function_dir)
-        service = Service(str(request.metadata.service_url), request.metadata.service_token)
+        service = Service(
+            str(request.metadata.service_url) if request.metadata.service_url else None, request.metadata.service_token
+        )
 
         response = function_callback(request.metadata, request.event, service)
 
         if response is None:
             return ""
 
-        if isinstance(response, ActionUnion):
+        if isinstance(response, ActionUnion):  # type: ignore  # MyPy doesn't like PEP604
             # wrap returned Actions into a WorkloadResponse
             response = WorkloadResponse(actions=[response])
-        elif isinstance(response, list) and all(isinstance(o, ActionUnion) for o in response):
+        elif isinstance(response, list) and all(isinstance(o, ActionUnion) for o in response):  # type: ignore  # MyPy doesn't like PEP604
             # wrap list of Actions into a WorkloadResponse
             response = WorkloadResponse(actions=response)
 
-        if not isinstance(
-            response, ResponseUnion
-        ):  # need to check for ResponseUnion instead of Response, because isinstance doesn't work with annotated unions
+        if not isinstance(response, ResponseUnion):  # type: ignore  # MyPy doesn't like PEP604
+            # need to check for ResponseUnion instead of Response, because isinstance doesn't work with annotated unions
             raise ValueError("Function needs to return a Response object or None.")
 
         # make sure the event_id is filled out correctly

csfunctions/metadata.py

@@ -1,31 +1,10 @@
 from datetime import datetime
+from typing import Optional
 
 from pydantic import AnyHttpUrl, BaseModel, Field
 
 
 class MetaData(BaseModel):
-    def __init__(
-        self,
-        app_lang: str,
-        app_user: str,
-        request_id: str,
-        request_datetime: datetime,
-        transaction_id: str,
-        instance_url: str,
-        db_service_url: str | None = None,
-        **kwargs,
-    ):
-        super().__init__(
-            app_lang=app_lang,
-            app_user=app_user,
-            request_id=request_id,
-            db_service_url=db_service_url,
-            request_datetime=request_datetime,
-            transaction_id=transaction_id,
-            instance_url=instance_url,
-            **kwargs,
-        )
-
     app_lang: str = Field(..., description="ISO code of the session language that triggered the webhook.")
     app_user: str = Field(..., description="User id of the user that triggered the webhook. (personalnummer)")
     request_id: str = Field(..., description="Unique identifier of this request.")
@@ -34,6 +13,6 @@ def __init__(
     request_datetime: datetime = Field(..., description="Time when the request was started.")
     transaction_id: str = Field(..., description="Unique identifier of the transaction.")
     instance_url: AnyHttpUrl = Field(..., description="URL to the instance where the webhook was triggered.")
-    db_service_url: AnyHttpUrl | None = Field(
+    db_service_url: Optional[AnyHttpUrl] = Field(
         None, description="URL to the DB Access Service responsible for the instance."
     )

csfunctions/objects/classification.py

@@ -16,11 +16,11 @@ class ObjectPropertyValue(BaseObject):
     ref_object_id: str = Field(..., description="Referenced Object")
     boolean_value: int | None = Field(..., description="Boolean Value")
     datetime_value: datetime | None = Field(..., description="Datetime Value")
-    float_value: float | None = Field("", description="Float Value")
-    float_value_normalized: float | None = Field("", description="Float Value Normalized")
-    integer_value: int | None = Field("", description="Integer Value")
+    float_value: float | None = Field(None, description="Float Value")
+    float_value_normalized: float | None = Field(None, description="Float Value Normalized")
+    integer_value: int | None = Field(None, description="Integer Value")
     iso_language_code: str | None = Field(None, description="ISO Language Code")
-    value_pos: int | None = Field("", description="Position")
+    value_pos: int | None = Field(None, description="Position")
     property_code: str | None = Field("", description="Property Code")
     property_path: str | None = Field(None, description="Property Path")
     property_type: str | None = Field(None, description="Property Type")

csfunctions/objects/document.py

@@ -98,4 +98,4 @@ class CADDocument(Document):
     Special Document type that contains a CAD-Model.
     """
 
-    object_type: Literal[ObjectType.CAD_DOCUMENT] = ObjectType.CAD_DOCUMENT
+    object_type: Literal[ObjectType.CAD_DOCUMENT] = ObjectType.CAD_DOCUMENT  # type: ignore[assignment]

csfunctions/objects/workflow.py

@@ -25,14 +25,14 @@ class Workflow(BaseObject):
     global_briefcases: list[Briefcase] = Field([], exclude=True)
 
     def link_objects(self, data: "EventData"):
-        local_briefcases = getattr(data, "local_briefcases", None)
-        global_briefcases = getattr(data, "local_briefcases", None)
+        local_briefcases: list[Briefcase] | None = getattr(data, "local_briefcases", None)
+        global_briefcases: list[Briefcase] | None = getattr(data, "local_briefcases", None)
 
         if local_briefcases and self.local_briefcase_ids:
             self._link_local_briefcases(local_briefcases)
 
         if global_briefcases and self.global_briefcase_ids:
-            self._link_global_briefcases(local_briefcases)
+            self._link_global_briefcases(global_briefcases)
 
     def _link_local_briefcases(self, local_briefcases: list["Briefcase"]):
         for local_briefcase in local_briefcases:

csfunctions/service.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 import requests
 
 
@@ -6,7 +8,7 @@ class Service:
     Provides access to services on the elements instance, e.g. generating numbers.
     """
 
-    def __init__(self, service_url: str, service_token: str):
+    def __init__(self, service_url: str | None, service_token: str | None):
         self.generator = NumberGeneratorService(service_url, service_token)
 
 
@@ -23,7 +25,7 @@ def __init__(self, service_url: str | None, service_token: str | None):
         self.service_url = service_url
         self.service_token = service_token
 
-    def request(self, endpoint: str, method: str = "GET", params: dict = None) -> dict | list:
+    def request(self, endpoint: str, method: str = "GET", params: Optional[dict] = None) -> dict | list:
         """
         Make a request to the access service.
         """
@@ -35,7 +37,7 @@ def request(self, endpoint: str, method: str = "GET", params: dict = None) -> di
         headers = {"Authorization": f"Bearer {self.service_token}"}
         params = params or {}
         url = self.service_url.rstrip("/") + "/" + endpoint.lstrip("/")
-        response = requests.request(method, url=url, params=params, headers=headers)
+        response = requests.request(method, url=url, params=params, headers=headers, timeout=10)
 
         if response.status_code == 401:
             raise Unauthorized
@@ -78,4 +80,8 @@ def get_numbers(self, name: str, count: int) -> list[int]:
         """
         params = {"name": name, "count": count}
         data = self.request(self.endpoint, params=params)
+        if not isinstance(data, dict):
+            raise ValueError(f"Access service returned invalid data. Expected dict, got {type(data)}")
+        if "numbers" not in data:
+            raise ValueError(f"Access service returned invalid data. Expected 'numbers' key, got {data.keys()}")
         return data["numbers"]

json_schemas/request.json

@@ -882,6 +882,7 @@
       "properties": {
         "name": {
           "const": "dummy",
+          "default": "dummy",
           "title": "Name",
           "type": "string"
         },
@@ -892,12 +893,12 @@
         },
         "data": {
           "$ref": "#/$defs/DummyEventData",
-          "default": []
+          "description": "Dummy Event Data"
         }
       },
       "required": [
-        "name",
-        "event_id"
+        "event_id",
+        "data"
       ],
       "title": "DummyEvent",
       "type": "object"
@@ -2666,6 +2667,7 @@
       "properties": {
         "name": {
           "const": "workflow_task_trigger",
+          "default": "workflow_task_trigger",
           "title": "Name",
           "type": "string"
         },
@@ -2679,7 +2681,6 @@
         }
       },
       "required": [
-        "name",
         "event_id",
         "data"
       ],

json_schemas/workload_response.json

@@ -35,6 +35,7 @@
       "properties": {
         "name": {
           "const": "dummy",
+          "default": "dummy",
           "title": "Name",
           "type": "string"
         },
@@ -51,9 +52,6 @@
           "title": "Id"
         }
       },
-      "required": [
-        "name"
-      ],
       "title": "DummyAction",
       "type": "object"
     }

tests/test_service.py

@@ -10,7 +10,7 @@ class TestNumberGeneratorService(TestCase):
     endpoint = "numgen"
     service_url = "https://some_service_url"
     service_token = "some_service_token"  # nosec
-    service = None
+    service: Service
 
     @classmethod
     def setUpClass(cls) -> None:

tests/test_workloadresponse.py

@@ -7,7 +7,7 @@
 
 
 class TestWorkloadResponse(TestCase):
-    def test_discriminator(self):
+    def test_discriminator(self) -> None:
         """
         Test that the discriminator on action objects (and responses) works
         """

tests/utils.py

@@ -3,6 +3,7 @@
 from csfunctions import DataResponse, MetaData, Request, Service
 from csfunctions.actions import AbortAndShowErrorAction
 from csfunctions.events import DummyEvent
+from csfunctions.events.dummy import DummyEventData
 from csfunctions.objects import Document, EngineeringChange, Part
 
 
@@ -35,8 +36,8 @@ def action_list_function(*args, **kwargs):  # pylint: disable=unused-argument
     return [AbortAndShowErrorAction(message="Testerror"), AbortAndShowErrorAction(message="Testerror")]
 
 
-dummy_document = Document(
-    **{
+dummy_document = Document.model_validate(
+    {
         "object_type": "document",
         "z_nummer": "D000017",
         "z_index": "a",
@@ -86,8 +87,8 @@ def action_list_function(*args, **kwargs):  # pylint: disable=unused-argument
     }
 )
 
-dummy_part = Part(
-    **{
+dummy_part = Part.model_validate(
+    {
         "object_type": "part",
         "teilenummer": "000000",
         "t_index": "a",
@@ -140,21 +141,25 @@ def action_list_function(*args, **kwargs):  # pylint: disable=unused-argument
 )
 
 dummy_request = Request(
-    metadata=MetaData(
-        request_id="123",
-        app_lang="de",
-        app_user="caddok",
-        request_datetime=datetime(2000, 1, 1),
-        transaction_id="123asd",
-        instance_url="https://instance.contact-cloud.com",
-        service_url=None,
+    metadata=MetaData.model_validate(
+        {
+            "request_id": "123",
+            "app_lang": "de",
+            "app_user": "caddok",
+            "request_datetime": datetime(2000, 1, 1),
+            "transaction_id": "123asd",
+            "instance_url": "https://instance.contact-cloud.com",
+            "service_url": None,
+            "service_token": "123",
+            "db_service_url": None,
+        }
     ),
-    event=DummyEvent(event_id="42", data={"documents": [dummy_document], "parts": [dummy_part]}),
+    event=DummyEvent(event_id="42", data=DummyEventData(documents=[dummy_document], parts=[dummy_part])),
 )
 
 
-dummy_ec = EngineeringChange(
-    **{
+dummy_ec = EngineeringChange.model_validate(
+    {
         "object_type": "engineering_change",
         "cdb_ec_id": "EC00000005",
         "cdb_project_id": "",