Use standard lib dataclasses instead of Pydantic

CHANGELOG.md

@@ -1,3 +1,8 @@
+## ?.?.? - 2024-03-18
+
+* Helpers for status data does no longer rely on Pydantic, but uses standard
+  dataclasses instead.
+
 ## 2.1.0 - 2024-02-15
 
 * New utility function `okdata.aws.ssm.get_secret` for retrieving secure strings

okdata/aws/status/model.py

@@ -1,7 +1,8 @@
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
 from enum import Enum
+from json import dumps, JSONEncoder
 from typing import Dict, Optional, List
-from datetime import datetime, timezone
-from pydantic import BaseModel, Field, validator
 
 
 class TraceStatus(str, Enum):
@@ -15,6 +16,35 @@ class TraceEventStatus(str, Enum):
     FAILED = "FAILED"
 
 
+class StatusJSONEncoder(JSONEncoder):
+    def default(self, obj, *args, **kwargs):
+        if isinstance(obj, datetime):
+            return obj.isoformat()
+        return super().default(obj)
+
+
+class BaseModel:
+    def dict(self, exclude_none=False):
+        if exclude_none:
+            return asdict(
+                self,
+                dict_factory=lambda d: {k: v for (k, v) in d if v is not None},
+            )
+        return asdict(self)
+
+    def json(self, exclude_none=False, **kwargs):
+        return dumps(
+            self.dict(exclude_none=exclude_none),
+            cls=StatusJSONEncoder,
+            **kwargs,
+        )
+
+    @classmethod
+    def parse_obj(cls, obj):
+        return cls(**obj)
+
+
+@dataclass
 class StatusMeta(BaseModel):
     function_name: Optional[str] = None
     function_version: Optional[str] = None
@@ -27,17 +57,14 @@ class StatusMeta(BaseModel):
 # TODO: Rework optional vs required and defaults when currents users are updated
 # and if to be used as a basis for the status api. Both trace_id (for new traces)
 # and trace_event_id are currently generated in status-api.
+@dataclass
 class StatusData(BaseModel):
     trace_id: Optional[str] = None  # TODO: Generate here as default?
     # trace_event_id: UUID = None  # = Field(default_factory=uuid4)
     domain: str = "N/A"  # TODO: Temporary default (required)
     domain_id: Optional[str] = None
-    start_time: datetime = Field(
-        default_factory=lambda: datetime.now(timezone.utc), str=datetime.isoformat
-    )
-    end_time: datetime = Field(
-        default_factory=lambda: datetime.now(timezone.utc), str=datetime.isoformat
-    )
+    start_time: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    end_time: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
     trace_status: TraceStatus = TraceStatus.CONTINUE
     trace_event_status: TraceEventStatus = TraceEventStatus.OK
     user: Optional[str] = None
@@ -50,23 +77,21 @@ class StatusData(BaseModel):
     exception: Optional[str] = None
     errors: Optional[List] = None
 
-    class Config:
-        validate_assignment = True
-
-    @validator("exception", pre=True)
-    def ensure_exception_data_is_string(cls, v):
-        if isinstance(v, Exception):
-            return str(v)
-        return v
-
-    @validator("errors", each_item=True)
-    def ensure_format_of_errors(cls, v):
-        if not isinstance(v, dict):
-            raise TypeError(f"{v} is not a dict.")
-        if "message" not in v:
-            raise ValueError("Missing key 'message'.")
-        if not isinstance(v["message"], dict):
-            raise TypeError("error['message'] is not a dict.")
-        if "nb" not in v["message"]:
-            raise ValueError("Missing key 'nb' in error['message'].")
-        return v
+    def __post_init__(self):
+        # Ensure that `meta` is of type `StatusMeta` if provided as a dictionary.
+        if isinstance(self.meta, dict):
+            self.meta = StatusMeta(**self.meta)
+
+        # Ensure that exception data is a string
+        self.exception = str(self.exception) if self.exception else None
+
+        # Validate and ensure format of errors
+        for error in self.errors or []:
+            if not isinstance(error, dict):
+                raise TypeError(f"{error} is not a dict.")
+            if "message" not in error:
+                raise ValueError("Missing key 'message'.")
+            if not isinstance(error["message"], dict):
+                raise TypeError("error['message'] is not a dict.")
+            if "nb" not in error["message"]:
+                raise ValueError("Missing key 'nb' in error['message'].")

requirements.txt

@@ -36,8 +36,6 @@ pyasn1==0.4.8
     # via
     #   python-jose
     #   rsa
-pydantic==1.8.2
-    # via okdata-aws (setup.py)
 pyjwt==2.4.0
     # via okdata-sdk
 pyrsistent==0.18.0
@@ -69,8 +67,6 @@ starlette==0.36.2
     # via okdata-aws (setup.py)
 structlog==21.2.0
     # via okdata-aws (setup.py)
-typing-extensions==3.10.0.2
-    # via pydantic
 urllib3==1.26.18
     # via
     #   botocore

setup.py

@@ -30,7 +30,6 @@
     install_requires=[
         "boto3",
         "okdata-sdk>=3,<4",
-        "pydantic<2",
         "requests",
         "starlette>=0.25.0,<1.0.0",
         "structlog",

tests/test_model.py

@@ -1,5 +1,4 @@
 import pytest
-from pydantic.error_wrappers import ValidationError
 
 from okdata.aws.status.model import StatusData
 
@@ -15,7 +14,7 @@ def test_errors_entry_valid(self):
 
     def test_errors_entry_not_dict(self):
         params = {"errors": [OK_ERROR, "This is string, not a dict."]}
-        with pytest.raises(ValidationError):
+        with pytest.raises(TypeError):
             StatusData(**params)
 
     def test_errors_entry_no_message(self):