Refactor to use async openai

* Don't block the event loop in OpenAI chat streaming

* Close OpenAI stream opened after the caller disconnected

* Show GenAI worker thread usage in system metrics

* Remove GenAI thread stats per review

* Use Starlette iterate_in_threadpool for OpenAI stream

frigate/api/chat.py

@@ -1,5 +1,6 @@
 """Chat and LLM tool calling APIs."""
 
+import asyncio
 import base64
 import json
 import logging
@@ -1653,7 +1654,8 @@ async def chat_completion(
                     f"Calling LLM (iteration {tool_iterations + 1}/{max_iterations}) "
                     f"with {len(conversation)} message(s) in conversation"
                 )
-                response = genai_client.chat_with_tools(
+                response = await asyncio.to_thread(
+                    genai_client.chat_with_tools,
                     messages=conversation,
                     tools=llm_tools if llm_tools else None,
                     tool_choice="auto",

frigate/data_processing/post/review_descriptions.py

@@ -34,7 +34,7 @@ from frigate.const import (
 from frigate.data_processing.types import PostProcessDataEnum
 from frigate.genai import GenAIClient
 from frigate.genai.manager import GenAIClientManager
-from frigate.models import Event, Recordings, ReviewSegment
+from frigate.models import Recordings, ReviewSegment
 from frigate.util.builtin import EventsPerSecond, InferenceSpeed
 from frigate.util.image import get_image_from_recording
 from frigate.util.recording_coverage import recording_overlap
@@ -439,22 +439,13 @@ class ReviewDescriptionProcessor(PostProcessorApi):
         if camera_config.review.genai.debug_save_thumbnails:
             self.save_debug_recording_frames(review_id, frames)
 
-        self.start_analysis(
-            camera_config,
-            final_data,
-            frames,
-            get_named_objects(
-                final_data["data"].get("detections") or [],
-                self.config.all_attributes,
-            ),
-        )
+        self.start_analysis(camera_config, final_data, frames)
 
     def start_analysis(
         self,
         camera_config: CameraConfig,
         final_data: dict[str, Any],
         frames: list[tuple[bytes, float]],
-        named_objects: list[tuple[str, str]] | None = None,
     ) -> None:
         """Kick off description generation for a review item in the background."""
         thumbs = [frame for frame, _ in frames]
@@ -490,7 +481,6 @@ class ReviewDescriptionProcessor(PostProcessorApi):
                 camera_config.review.genai,
                 sorted(self.config.all_labels),
                 self.config.all_attributes,
-                named_objects,
             ),
         ).start()
 
@@ -751,29 +741,6 @@ def format_classification_state_changes(
     return lines
 
 
-def get_named_objects(
-    detection_ids: list[str], attribute_labels: list[str]
-) -> list[tuple[str, str]]:
-    """Get the label and sub label of each named tracked object in a review item.
-
-    Editing a sub label after a review item ends rebuilds the stored sub_labels
-    without verified_objects, so the stored lists can't be paired by position.
-    """
-    events = (
-        Event.select(Event.label, Event.sub_label)
-        .where(Event.id.in_(detection_ids), Event.sub_label.is_null(False))
-        .order_by(Event.start_time)
-    )
-
-    # attribute sub labels (eg, delivery logos) are not names, which matches
-    # how the review maintainer records them
-    return [
-        (event.label, event.sub_label)
-        for event in events
-        if event.sub_label and event.sub_label not in attribute_labels
-    ]
-
-
 def run_analysis(
     requestor: InterProcessRequestor,
     genai_client: GenAIClient,
@@ -785,7 +752,6 @@ def run_analysis(
     genai_config: GenAIReviewConfig,
     labelmap_objects: list[str],
     attribute_labels: list[str],
-    named_objects: list[tuple[str, str]] | None,
 ) -> None:
     start = datetime.datetime.now().timestamp()
 
@@ -810,20 +776,11 @@ def run_analysis(
     unified_objects = []
 
     objects_list = final_data["data"]["objects"]
+    sub_labels_list = final_data["data"]["sub_labels"]
 
-    if named_objects is None:
-        # live review items record a sub label for every verified object
-        named_objects = [
-            (verified_label.replace("-verified", ""), sub_label)
-            for verified_label, sub_label in zip(
-                final_data["data"]["verified_objects"],
-                final_data["data"]["sub_labels"],
-            )
-        ]
-
-    for label, sub_label in named_objects:
-        object_type = label.replace("_", " ")
-        name = titlecase(sub_label.replace("_", " "))
+    for i, verified_label in enumerate(final_data["data"]["verified_objects"]):
+        object_type = verified_label.replace("-verified", "").replace("_", " ")
+        name = titlecase(sub_labels_list[i].replace("_", " "))
         unified_objects.append(f"{name} ← {object_type}")
 
     for label in objects_list:

frigate/genai/plugins/azure-openai.py

@@ -7,7 +7,8 @@ overrides what is genuinely Azure-specific:
 
 - Client construction: parses ``api-version`` out of the configured
   ``base_url`` query string and instantiates :class:`openai.AzureOpenAI`
-  with ``azure_endpoint`` instead of ``base_url``. Raises if the URL is
+  (or :class:`openai.AsyncAzureOpenAI` for streaming chat) with
+  ``azure_endpoint`` instead of ``base_url``. Raises if the URL is
   malformed; :class:`GenAIClientManager` catches the exception and
   disables the provider.
 - Context size: Azure does not expose a per-model ``max_model_len`` field
@@ -29,9 +30,10 @@ the ``transcribe`` role:
 """
 
 import logging
+from typing import Any
 from urllib.parse import parse_qs, urlparse
 
-from openai import AzureOpenAI
+from openai import AsyncAzureOpenAI, AzureOpenAI
 
 from frigate.config import GenAIProviderEnum
 from frigate.genai import register_genai_provider
@@ -44,8 +46,8 @@ logger = logging.getLogger(__name__)
 class AzureOpenAIClient(OpenAIClient):
     """Generative AI client for Frigate using Azure OpenAI."""
 
-    def _init_provider(self) -> AzureOpenAI:
-        """Initialize the AzureOpenAI client from the configured base_url."""
+    def _client_options(self) -> dict[str, Any]:
+        """Build the Azure client options from the configured base_url."""
         parsed_url = urlparse(self.genai_config.base_url or "")
         query_params = parse_qs(parsed_url.query)
         api_version = query_params.get("api-version", [None])[0]
@@ -53,13 +55,19 @@ class AzureOpenAIClient(OpenAIClient):
         if not api_version:
             raise ValueError("Azure OpenAI base_url is missing api-version.")
 
-        azure_endpoint = f"{parsed_url.scheme}://{parsed_url.netloc}/"
+        return {
+            "api_key": self.genai_config.api_key,
+            "api_version": api_version,
+            "azure_endpoint": f"{parsed_url.scheme}://{parsed_url.netloc}/",
+        }
 
-        return AzureOpenAI(
-            api_key=self.genai_config.api_key,
-            api_version=api_version,
-            azure_endpoint=azure_endpoint,
-        )
+    def _init_provider(self) -> AzureOpenAI:
+        """Initialize the AzureOpenAI client from the configured base_url."""
+        return AzureOpenAI(**self._client_options())
+
+    def _init_async_provider(self) -> AsyncAzureOpenAI:
+        """Create an async Azure client for streaming chat."""
+        return AsyncAzureOpenAI(**self._client_options())
 
     def get_context_size(self) -> int:
         """Azure does not reliably surface per-model context size; use 128K."""

frigate/genai/plugins/openai.py

@@ -7,7 +7,7 @@ from collections.abc import AsyncGenerator
 from typing import Any
 
 from httpx import TimeoutException
-from openai import OpenAI
+from openai import AsyncOpenAI, OpenAI
 
 from frigate.config import GenAIProviderEnum
 from frigate.genai import GenAIClient, register_genai_provider
@@ -15,6 +15,7 @@ from frigate.genai.utils import interleave_images
 
 logger = logging.getLogger(__name__)
 
+
 # gpt-transcribe replaced the singular `language` field with a `languages` array
 # and rejects a request that sends both. Older transcription models
 # (gpt-4o-transcribe, gpt-4o-mini-transcribe, whisper-1) still take the singular
@@ -45,8 +46,8 @@ class OpenAIClient(GenAIClient):
     provider: OpenAI
     context_size: int | None = None
 
-    def _init_provider(self) -> OpenAI:
-        """Initialize the client.
+    def _client_options(self) -> dict[str, Any]:
+        """Build the constructor options shared by the sync and async clients.
 
         Subclasses (e.g. Azure) should raise on configuration errors; the
         manager catches construction failures and disables the provider.
@@ -62,7 +63,19 @@ class OpenAIClient(GenAIClient):
         if self.genai_config.base_url:
             provider_opts["base_url"] = self.genai_config.base_url
 
-        return OpenAI(api_key=self.genai_config.api_key, **provider_opts)
+        return {"api_key": self.genai_config.api_key, **provider_opts}
+
+    def _init_provider(self) -> OpenAI:
+        """Initialize the client."""
+        return OpenAI(**self._client_options())
+
+    def _init_async_provider(self) -> AsyncOpenAI:
+        """Create an async client for streaming chat.
+
+        A new client is created per stream because its connection pool binds
+        to the event loop that first uses it.
+        """
+        return AsyncOpenAI(**self._client_options())
 
     def _send(
         self,
@@ -412,57 +425,61 @@ class OpenAIClient(GenAIClient):
             finish_reason = "stop"
             usage_stats: dict[str, Any] | None = None
 
-            stream = self.provider.chat.completions.create(**request_params)
+            # Exiting these blocks, including when the caller disconnects and
+            # the generator is cancelled, closes the response and its connection
+            async with (
+                self._init_async_provider() as client,
+                await client.chat.completions.create(**request_params) as stream,
+            ):
+                async for chunk in stream:
+                    chunk_usage = getattr(chunk, "usage", None)
+                    if chunk_usage is not None:
+                        usage_stats = _stats_from_openai_usage(chunk_usage)
 
-            for chunk in stream:
-                chunk_usage = getattr(chunk, "usage", None)
-                if chunk_usage is not None:
-                    usage_stats = _stats_from_openai_usage(chunk_usage)
+                    if not chunk or not chunk.choices:
+                        continue
 
-                if not chunk or not chunk.choices:
-                    continue
+                    choice = chunk.choices[0]
+                    delta = choice.delta
 
-                choice = chunk.choices[0]
-                delta = choice.delta
+                    # Check for finish reason
+                    if choice.finish_reason:
+                        finish_reason = choice.finish_reason
 
-                # Check for finish reason
-                if choice.finish_reason:
-                    finish_reason = choice.finish_reason
+                    # Extract reasoning deltas (reasoning_content or reasoning,
+                    # depending on the server)
+                    reasoning_delta = getattr(
+                        delta, "reasoning_content", None
+                    ) or getattr(delta, "reasoning", None)
+                    if reasoning_delta:
+                        reasoning_parts.append(reasoning_delta)
+                        yield ("reasoning_delta", reasoning_delta)
 
-                # Extract reasoning deltas (reasoning_content or reasoning,
-                # depending on the server)
-                reasoning_delta = getattr(delta, "reasoning_content", None) or getattr(
-                    delta, "reasoning", None
-                )
-                if reasoning_delta:
-                    reasoning_parts.append(reasoning_delta)
-                    yield ("reasoning_delta", reasoning_delta)
+                    # Extract content deltas
+                    if delta.content:
+                        content_parts.append(delta.content)
+                        yield ("content_delta", delta.content)
 
-                # Extract content deltas
-                if delta.content:
-                    content_parts.append(delta.content)
-                    yield ("content_delta", delta.content)
+                    # Extract tool calls
+                    if delta.tool_calls:
+                        for tc in delta.tool_calls:
+                            idx = tc.index
+                            fn = tc.function
 
-                # Extract tool calls
-                if delta.tool_calls:
-                    for tc in delta.tool_calls:
-                        idx = tc.index
-                        fn = tc.function
+                            if idx not in tool_calls_by_index:
+                                tool_calls_by_index[idx] = {
+                                    "id": tc.id or "",
+                                    "name": fn.name if fn and fn.name else "",
+                                    "arguments": "",
+                                }
 
-                        if idx not in tool_calls_by_index:
-                            tool_calls_by_index[idx] = {
-                                "id": tc.id or "",
-                                "name": fn.name if fn and fn.name else "",
-                                "arguments": "",
-                            }
-
-                        t = tool_calls_by_index[idx]
-                        if tc.id:
-                            t["id"] = tc.id
-                        if fn and fn.name:
-                            t["name"] = fn.name
-                        if fn and fn.arguments:
-                            t["arguments"] += fn.arguments
+                            t = tool_calls_by_index[idx]
+                            if tc.id:
+                                t["id"] = tc.id
+                            if fn and fn.name:
+                                t["name"] = fn.name
+                            if fn and fn.arguments:
+                                t["arguments"] += fn.arguments
 
             # Build final message
             full_content = "".join(content_parts).strip() or None

frigate/test/test_genai_providers.py

@@ -145,12 +145,72 @@ def _openai_chunk(content=None, tool_calls=None, finish_reason=None, usage=None)
     return SimpleNamespace(choices=[choice], usage=usage)
 
 
+class _FakeAsyncOpenAIStream:
+    """Mimics openai.AsyncStream: async iterable and async context manager.
+
+    With ``stall_after`` set, the stream stops producing after that many
+    chunks, like a model that stops sending tokens, and sets ``stalled``.
+    """
+
+    def __init__(self, chunks, stall_after=None):
+        self._chunks = chunks
+        self._stall_after = stall_after
+        self.stalled = asyncio.Event()
+        self.close = AsyncMock()
+
+    async def __aiter__(self):
+        for i, chunk in enumerate(self._chunks):
+            if i == self._stall_after:
+                self.stalled.set()
+                await asyncio.Event().wait()
+            yield chunk
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        await self.close()
+
+
+class _FakeAsyncOpenAI:
+    """Mimics openai.AsyncOpenAI: async context manager with a completions API."""
+
+    def __init__(self, create):
+        self.chat = SimpleNamespace(completions=SimpleNamespace(create=create))
+        self.close = AsyncMock()
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        await self.close()
+
+
 class TestOpenAIProvider(unittest.TestCase):
     def _client(self):
         return _make_client(
             "openai", model="gpt-4o", api_key="k", base_url="http://localhost:9999/v1"
         )
 
+    def _streaming_client(self, create):
+        """Build a client whose async completions call is ``create``."""
+        client = self._client()
+        async_client = _FakeAsyncOpenAI(create)
+        client._init_async_provider = MagicMock(return_value=async_client)
+        return client, async_client
+
+    def test_async_client_uses_configured_options(self):
+        from openai import AsyncOpenAI
+
+        client = self._client()
+        async_client = client._init_async_provider()
+        try:
+            self.assertIsInstance(async_client, AsyncOpenAI)
+            self.assertEqual(async_client.base_url, client.provider.base_url)
+            self.assertEqual(async_client.api_key, "k")
+        finally:
+            asyncio.run(async_client.close())
+
     def test_stream_tool_call_arguments_are_dict(self):
         # Arguments arrive split across chunks, as the real API streams them.
         chunks = [
@@ -162,8 +222,9 @@ class TestOpenAIProvider(unittest.TestCase):
             _openai_chunk(tool_calls=[_openai_tc(0, arguments=' "person"}')]),
             _openai_chunk(finish_reason="tool_calls"),
         ]
-        client = self._client()
-        client.provider.chat.completions.create = MagicMock(return_value=iter(chunks))
+        client, _ = self._streaming_client(
+            AsyncMock(return_value=_FakeAsyncOpenAIStream(chunks))
+        )
 
         final = _final_message(_collect(client, SIMPLE_MESSAGES))
         self.assertEqual(final["finish_reason"], "tool_calls")
@@ -177,8 +238,9 @@ class TestOpenAIProvider(unittest.TestCase):
             _openai_chunk(content="lo"),
             _openai_chunk(finish_reason="stop"),
         ]
-        client = self._client()
-        client.provider.chat.completions.create = MagicMock(return_value=iter(chunks))
+        client, _ = self._streaming_client(
+            AsyncMock(return_value=_FakeAsyncOpenAIStream(chunks))
+        )
 
         events = _collect(client, SIMPLE_MESSAGES)
         deltas = [v for (k, v) in events if k == "content_delta"]
@@ -186,14 +248,84 @@ class TestOpenAIProvider(unittest.TestCase):
         self.assertEqual(_final_message(events)["content"], "hello")
 
     def test_multimodal_message_does_not_crash(self):
-        client = self._client()
-        client.provider.chat.completions.create = MagicMock(
-            return_value=iter([_openai_chunk(content="ok", finish_reason="stop")])
+        client, _ = self._streaming_client(
+            AsyncMock(
+                return_value=_FakeAsyncOpenAIStream(
+                    [_openai_chunk(content="ok", finish_reason="stop")]
+                )
+            )
         )
         # Passing the OpenAI-native multimodal list through must not raise.
         final = _final_message(_collect(client, MULTIMODAL_MESSAGES))
         self.assertEqual(final["content"], "ok")
 
+    def test_stream_uses_async_client_and_closes_it(self):
+        # The blocking sync client must never be used on the event loop.
+        stream = _FakeAsyncOpenAIStream(
+            [_openai_chunk(content="ok", finish_reason="stop")]
+        )
+        create = AsyncMock(return_value=stream)
+        client, async_client = self._streaming_client(create)
+        client.provider.chat.completions.create = MagicMock()
+
+        final = _final_message(_collect(client, SIMPLE_MESSAGES))
+
+        self.assertEqual(final["content"], "ok")
+        self.assertTrue(create.await_args.kwargs["stream"])
+        client.provider.chat.completions.create.assert_not_called()
+        stream.close.assert_awaited_once()
+        async_client.close.assert_awaited_once()
+
+    def test_client_closed_when_caller_leaves_while_opening(self):
+        # Cancelling before the response arrives must still close the client.
+        async def _run():
+            opening = asyncio.Event()
+
+            async def hang(**kwargs):
+                opening.set()
+                await asyncio.Event().wait()
+
+            client, async_client = self._streaming_client(hang)
+            gen = client.chat_with_tools_stream(
+                messages=SIMPLE_MESSAGES, tools=None, tool_choice="auto"
+            )
+            task = asyncio.create_task(gen.__anext__())
+            await opening.wait()
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await asyncio.wait_for(task, timeout=1)
+            return async_client
+
+        async_client = asyncio.run(_run())
+        async_client.close.assert_awaited_once()
+
+    def test_stream_closed_when_caller_leaves_mid_stream(self):
+        # Cancelling while waiting on a stalled model must close the stream
+        # right away instead of waiting for the next chunk.
+        async def _run():
+            stream = _FakeAsyncOpenAIStream(
+                [_openai_chunk(content="a"), _openai_chunk(finish_reason="stop")],
+                stall_after=1,
+            )
+            client, async_client = self._streaming_client(
+                AsyncMock(return_value=stream)
+            )
+            gen = client.chat_with_tools_stream(
+                messages=SIMPLE_MESSAGES, tools=None, tool_choice="auto"
+            )
+            first = await gen.__anext__()
+            task = asyncio.create_task(gen.__anext__())
+            await stream.stalled.wait()
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await asyncio.wait_for(task, timeout=1)
+            return first, stream, async_client
+
+        first, stream, async_client = asyncio.run(_run())
+        self.assertEqual(first, ("content_delta", "a"))
+        stream.close.assert_awaited_once()
+        async_client.close.assert_awaited_once()
+
 
 # ---------------------------------------------------------------------------
 # Gemini
@@ -940,6 +1072,17 @@ class TestAzureOpenAITranscribe(unittest.TestCase):
         self.assertIsInstance(client.provider, AzureOpenAI)
         self.assertTrue(client.supports_transcription)
 
+    def test_streaming_routes_through_async_azure_client(self):
+        from openai import AsyncAzureOpenAI
+
+        client = self._client()
+        async_client = client._init_async_provider()
+        try:
+            self.assertIsInstance(async_client, AsyncAzureOpenAI)
+            self.assertEqual(async_client.base_url, client.provider.base_url)
+        finally:
+            asyncio.run(async_client.close())
+
     def test_transcribe_inherited(self):
         client = self._client()
         create = MagicMock(return_value="azure text")

frigate/test/test_review_sub_label_sync.py

@@ -1,159 +0,0 @@
-"""Tests for pairing review sub labels with their verified objects."""
-
-import unittest
-from unittest.mock import MagicMock
-
-from peewee import SqliteDatabase
-
-from frigate.config import FrigateConfig
-from frigate.data_processing.post.review_descriptions import (
-    get_named_objects,
-    run_analysis,
-)
-from frigate.models import Event
-
-CONFIG = """
-mqtt:
-  enabled: False
-cameras:
-  front_door:
-    ffmpeg:
-      inputs:
-        - path: rtsp://10.0.0.1:554/video
-          roles:
-            - detect
-    detect:
-      width: 1920
-      height: 1080
-      fps: 5
-"""
-
-
-def unified_objects(
-    review_data: dict, named_objects: list[tuple[str, str]] | None = None
-) -> list[str]:
-    """Run the review analysis and return the objects sent to the GenAI client."""
-    camera_config = FrigateConfig.parse_yaml(CONFIG).cameras["front_door"]
-    client = MagicMock()
-    client.generate_review_description.return_value = None
-    run_analysis(
-        MagicMock(),
-        client,
-        MagicMock(),
-        camera_config,
-        {"id": "review1", "start_time": 0, "end_time": 10, "data": review_data},
-        [],
-        [],
-        camera_config.review.genai,
-        ["car", "person"],
-        ["amazon"],
-        named_objects,
-    )
-    return client.generate_review_description.call_args.args[0]["unified_objects"]
-
-
-class TestRunAnalysisSubLabels(unittest.TestCase):
-    def test_verified_objects_are_named_by_their_sub_label(self):
-        objects = unified_objects(
-            {
-                "zones": [],
-                "objects": ["person-verified", "car"],
-                "verified_objects": ["person-verified"],
-                "sub_labels": ["nick_m"],
-                "classification_state_changes": [],
-            }
-        )
-
-        self.assertEqual(objects, ["Nick M ← person", "car"])
-
-
-class TestRegeneratedSubLabels(unittest.TestCase):
-    """Review items whose stored lists were rebuilt after a sub label edit."""
-
-    def setUp(self):
-        self.db = SqliteDatabase(":memory:")
-        self.db.bind([Event])
-        self.db.create_tables([Event])
-
-    def tearDown(self):
-        self.db.close()
-
-    def _insert_event(
-        self, id: str, label: str, sub_label: str | None, start_time: float
-    ) -> None:
-        Event.create(
-            id=id,
-            label=label,
-            sub_label=sub_label,
-            camera="front_door",
-            start_time=start_time,
-            end_time=start_time + 10,
-            top_score=0.9,
-            score=0.9,
-            false_positive=False,
-            zones=[],
-            thumbnail="",
-            has_clip=True,
-            has_snapshot=True,
-            region=[],
-            box=[],
-            area=0,
-            retain_indefinitely=False,
-            plus_id="",
-            model_hash="",
-            detector_type="cpu",
-            model_type="ssd",
-            data={},
-        )
-
-    def test_cleared_sub_label_does_not_move_to_another_object(self):
-        self._insert_event("car1", "car", None, 1)
-        self._insert_event("person1", "person", "Nick", 2)
-
-        named = get_named_objects(["car1", "person1"], ["amazon"])
-        objects = unified_objects(
-            {
-                "zones": [],
-                "objects": ["car", "person-verified"],
-                "verified_objects": ["car-verified", "person-verified"],
-                "sub_labels": ["Nick"],
-                "classification_state_changes": [],
-            },
-            named,
-        )
-
-        self.assertEqual(objects, ["Nick ← person", "car"])
-
-    def test_repeated_name_is_kept_for_each_object(self):
-        # sub_labels are deduplicated when rebuilt, which previously raised an
-        # IndexError when the verified objects outnumbered them
-        self._insert_event("person1", "person", "Nick", 1)
-        self._insert_event("person2", "person", "Nick", 2)
-
-        named = get_named_objects(["person1", "person2"], ["amazon"])
-        objects = unified_objects(
-            {
-                "zones": [],
-                "objects": ["person-verified", "person-verified"],
-                "verified_objects": ["person-verified", "person-verified"],
-                "sub_labels": ["Nick"],
-                "classification_state_changes": [],
-            },
-            named,
-        )
-
-        self.assertEqual(objects, ["Nick ← person", "Nick ← person"])
-
-    def test_attribute_sub_labels_are_not_names(self):
-        self._insert_event("car1", "car", "amazon", 1)
-        self._insert_event("person1", "person", "Nick", 2)
-        self._insert_event("other", "person", "Bob", 3)
-
-        self.assertEqual(
-            get_named_objects(["car1", "person1"], ["amazon"]),
-            [("person", "Nick")],
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()