add mypy

Author: Jens Kürten

.pre-commit-config.yaml

@@ -11,7 +11,16 @@ 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
+            ]
   - 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/handler.py

@@ -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
 
 
@@ -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/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"]

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
 
 
@@ -149,7 +150,7 @@ def action_list_function(*args, **kwargs):  # pylint: disable=unused-argument
         instance_url="https://instance.contact-cloud.com",
         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])),
 )