Feat: Add MyPy support (#20)

* add mypy

* mark package as typed

---------

Co-authored-by: Jens Kürten <[email protected]>

Author: jens-kuerten

.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: