From bf84f8558d40af7b807c749798a5bbb656ca4a4b Mon Sep 17 00:00:00 2001 From: Petter Holt Juliussen Date: Mon, 18 Mar 2024 11:57:52 +0100 Subject: [PATCH] Use standard lib dataclasses instead of Pydantic --- CHANGELOG.md | 5 +++ okdata/aws/status/model.py | 81 +++++++++++++++++++++++++------------- requirements.txt | 4 -- setup.py | 1 - tests/test_model.py | 3 +- 5 files changed, 59 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 786822d..847db29 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/okdata/aws/status/model.py b/okdata/aws/status/model.py index d276f02..41872a5 100644 --- a/okdata/aws/status/model.py +++ b/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'].") diff --git a/requirements.txt b/requirements.txt index 884b5ff..05fa0f1 100644 --- a/requirements.txt +++ b/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 diff --git a/setup.py b/setup.py index d5de17c..c112945 100644 --- a/setup.py +++ b/setup.py @@ -30,7 +30,6 @@ install_requires=[ "boto3", "okdata-sdk>=3,<4", - "pydantic<2", "requests", "starlette>=0.25.0,<1.0.0", "structlog", diff --git a/tests/test_model.py b/tests/test_model.py index 58b5409..e7bbac7 100644 --- a/tests/test_model.py +++ b/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):