Fix Pydantic deprecation warnings (#764)

Author: grahamalama

jbi/models.py

@@ -11,7 +11,14 @@
 from typing import DefaultDict, Literal, Mapping, Optional
 from urllib.parse import ParseResult, urlparse
 
-from pydantic import BaseModel, ConfigDict, Extra, Field, RootModel, field_validator
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    RootModel,
+    TypeAdapter,
+    field_validator,
+)
 from typing_extensions import TypedDict
 
 from jbi import Operation, steps
@@ -295,6 +302,9 @@ class BugzillaComment(BaseModel):
     creator: str
 
 
+BugzillaComments = TypeAdapter(list[BugzillaComment])
+
+
 class BugzillaApiResponse(BaseModel):
     """Bugzilla Response Object"""
 
@@ -335,7 +345,7 @@ class Context(BaseModel):
 
     def update(self, **kwargs):
         """Return a copy with updated fields."""
-        return self.copy(update=kwargs)
+        return self.model_copy(update=kwargs, deep=True)
 
 
 class JiraContext(Context):
@@ -349,7 +359,7 @@ class JiraContext(Context):
 BugId = TypedDict("BugId", {"id": Optional[int]})
 
 
-class RunnerContext(Context, extra=Extra.forbid):
+class RunnerContext(Context, extra="forbid"):
     """Logging context from runner"""
 
     rid: str
@@ -359,7 +369,7 @@ class RunnerContext(Context, extra=Extra.forbid):
     bug: BugId | BugzillaBug
 
 
-class ActionContext(Context, extra=Extra.forbid):
+class ActionContext(Context, extra="forbid"):
     """Logging context from actions"""
 
     action: Action

jbi/services/bugzilla.py

@@ -6,7 +6,6 @@
 from functools import lru_cache
 
 import requests
-from pydantic import parse_obj_as
 from statsd.defaults.env import statsd
 
 from jbi import Operation, environment
@@ -15,6 +14,7 @@
     BugzillaApiResponse,
     BugzillaBug,
     BugzillaComment,
+    BugzillaComments,
     BugzillaWebhooksResponse,
 )
 
@@ -73,7 +73,7 @@ def get_bug(self, bugid) -> BugzillaBug:
         # https://bugzilla.readthedocs.io/en/latest/api/core/v1/bug.html#rest-single-bug
         url = f"{self.base_url}/rest/bug/{bugid}"
         bug_info = self._call("GET", url)
-        parsed = BugzillaApiResponse.parse_obj(bug_info)
+        parsed = BugzillaApiResponse.model_validate(bug_info)
         if not parsed.bugs:
             raise BugzillaClientError(
                 f"Unexpected response content from 'GET {url}' (no 'bugs' field)"
@@ -85,7 +85,7 @@ def get_bug(self, bugid) -> BugzillaBug:
             matching_comments = [c for c in comment_list if c.id == bug.comment.id]
             # If no matching entry is found, set `bug.comment` to `None`.
             found = matching_comments[0] if matching_comments else None
-            bug = bug.copy(update={"comment": found})
+            bug = bug.model_copy(update={"comment": found}, deep=True)
         return bug
 
     @instrumented_method
@@ -99,15 +99,15 @@ def get_comments(self, bugid) -> list[BugzillaComment]:
             raise BugzillaClientError(
                 f"Unexpected response content from 'GET {url}' (no 'bugs' field)"
             )
-        return parse_obj_as(list[BugzillaComment], comments)
+        return BugzillaComments.validate_python(comments)
 
     @instrumented_method
     def update_bug(self, bugid, **fields) -> BugzillaBug:
         """Update the specified fields of the specified bug."""
         # https://bugzilla.readthedocs.io/en/latest/api/core/v1/bug.html#rest-update-bug
         url = f"{self.base_url}/rest/bug/{bugid}"
         updated_info = self._call("PUT", url, json=fields)
-        parsed = BugzillaApiResponse.parse_obj(updated_info)
+        parsed = BugzillaApiResponse.model_validate(updated_info)
         if not parsed.bugs:
             raise BugzillaClientError(
                 f"Unexpected response content from 'PUT {url}' (no 'bugs' field)"
@@ -119,7 +119,7 @@ def list_webhooks(self):
         """List the currently configured webhooks, including their status."""
         url = f"{self.base_url}/rest/webhooks/list"
         webhooks_info = self._call("GET", url)
-        parsed = BugzillaWebhooksResponse.parse_obj(webhooks_info)
+        parsed = BugzillaWebhooksResponse.model_validate(webhooks_info)
         if parsed.webhooks is None:
             raise BugzillaClientError(
                 f"Unexpected response content from 'GET {url}' (no 'webhooks' field)"

tests/unit/test_app.py

@@ -72,7 +72,7 @@ def test_errors_are_reported_to_sentry(
         with patch("jbi.router.execute_action", side_effect=ValueError):
             with pytest.raises(ValueError):
                 anon_client.post(
-                    "/bugzilla_webhook", data=webhook_create_example.json()
+                    "/bugzilla_webhook", data=webhook_create_example.model_dump_json()
                 )
 
     assert mocked.called, "Sentry captured the exception"
@@ -92,7 +92,7 @@ def test_request_id_is_passed_down_to_logger_contexts(
         with TestClient(app) as anon_client:
             anon_client.post(
                 "/bugzilla_webhook",
-                data=webhook_create_example.json(),
+                data=webhook_create_example.model_dump_json(),
                 headers={
                     "X-Request-Id": "foo-bar",
                 },

tests/unit/test_models.py

@@ -37,8 +37,8 @@ def test_default_invalid_step():
 
 def test_duplicated_whiteboard_tag_fails(action_factory):
     with pytest.raises(ValueError) as exc_info:
-        Actions.parse_obj(
-            [
+        Actions(
+            root=[
                 action_factory(whiteboard_tag="x"),
                 action_factory(whiteboard_tag="y"),
                 action_factory(whiteboard_tag="x"),
@@ -49,8 +49,8 @@ def test_duplicated_whiteboard_tag_fails(action_factory):
 
 def test_override_step_configuration_for_single_action_type():
     default_steps = ActionSteps()
-    params = ActionParams.parse_obj(
-        {"jira_project_key": "JBI", "steps": {"new": ["create_issue"]}}
+    params = ActionParams(
+        jira_project_key="JBI", steps=ActionSteps(new=["create_issue"])
     )
     assert params.steps.new == ["create_issue"]
     assert params.steps.new != default_steps.new

tests/unit/test_router.py

@@ -108,7 +108,7 @@ def test_webhook_is_200_if_action_succeeds(
 
     with TestClient(app) as anon_client:
         response = anon_client.post(
-            "/bugzilla_webhook", data=webhook_create_example.json()
+            "/bugzilla_webhook", data=webhook_create_example.model_dump_json()
         )
         assert response
         assert response.status_code == 200
@@ -124,7 +124,7 @@ def test_webhook_is_200_if_action_raises_IgnoreInvalidRequestError(
 
     with TestClient(app) as anon_client:
         response = anon_client.post(
-            "/bugzilla_webhook", data=webhook_create_example.json()
+            "/bugzilla_webhook", data=webhook_create_example.model_dump_json()
         )
         assert response
         assert response.status_code == 200