Replace blocking I/O in async functions

Replaces aiofiles with anyio, because anyio.Path is much more complete
and comparable to the Pathlib API.

docker/main/requirements-wheels.txt

@@ -1,4 +1,4 @@
-aiofiles == 24.1.*
+anyio == 4.14.*
 click == 8.1.*
 # FastAPI
 aiohttp == 3.12.*

frigate/api/app.py

@@ -14,8 +14,8 @@ from io import StringIO
 from pathlib import Path as FilePath
 from typing import Any
 
-import aiofiles
 import ruamel.yaml
+from anyio import open_file as aopen
 from fastapi import APIRouter, Body, Path, Request, Response
 from fastapi.encoders import jsonable_encoder
 from fastapi.params import Depends
@@ -1052,7 +1052,7 @@ async def logs(
         """Asynchronously stream log lines."""
         buffer = ""
         try:
-            async with aiofiles.open(file_path) as file:
+            async with await aopen(file_path) as file:
                 await file.seek(0, 2)
                 while True:
                     line = await file.readline()
@@ -1090,7 +1090,7 @@ async def logs(
 
     # For full logs initially
     try:
-        async with aiofiles.open(service_location) as file:
+        async with await aopen(service_location) as file:
             contents = await file.read()
 
         total_lines, log_lines = process_logs(contents, service, start, end)

frigate/api/camera.py

@@ -10,6 +10,7 @@ from urllib.parse import quote_plus
 
 import httpx
 import requests
+from anyio import open_file as aopen
 from fastapi import APIRouter, Depends, Query, Request, Response
 from fastapi.responses import JSONResponse
 from filelock import FileLock, Timeout
@@ -1187,15 +1188,17 @@ async def delete_camera(
 
     try:
         with lock:
-            with open(config_file) as f:
+            async with await aopen(config_file) as f:
-                old_raw_config = f.read()
+                old_raw_config = await f.read()
 
             try:
                 yaml = YAML()
                 yaml.indent(mapping=2, sequence=4, offset=2)
 
-                with open(config_file) as f:
+                async with await aopen(config_file) as f:
-                    data = yaml.load(f)
+                    text = await f.read()
+
+                data = yaml.load(text)
 
                 # Remove camera from config
                 if "cameras" in data and camera_name in data["cameras"]:
@@ -1220,17 +1223,17 @@ async def delete_camera(
                     for role_name in empty_roles:
                         del auth["roles"][role_name]
 
-                with open(config_file, "w") as f:
+                async with await aopen(config_file, "w") as f:
                     yaml.dump(data, f)
 
-                with open(config_file) as f:
+                async with await aopen(config_file) as f:
-                    new_raw_config = f.read()
+                    new_raw_config = await f.read()
 
                 try:
                     config = FrigateConfig.parse(new_raw_config)
                 except Exception:
-                    with open(config_file, "w") as f:
+                    async with await aopen(config_file, "w") as f:
-                        f.write(old_raw_config)
+                        await f.write(old_raw_config)
                     logger.exception(
                         "Config error after removing camera %s",
                         camera_name,

frigate/api/event.py

@@ -13,6 +13,7 @@ from pathlib import Path
 from urllib.parse import unquote
 
 import numpy as np
+from anyio import Path as AsyncPath
 from fastapi import APIRouter, Request
 from fastapi.params import Depends
 from fastapi.responses import JSONResponse
@@ -1455,10 +1456,10 @@ async def set_attributes(
         dataset_dir = os.path.join(CLIPS_DIR, sanitize_filename(model_key), "dataset")
         available_labels = set()
 
-        if os.path.exists(dataset_dir):
+        if await AsyncPath(dataset_dir).exists():
             for category_name in os.listdir(dataset_dir):
                 category_dir = os.path.join(dataset_dir, category_name)
-                if os.path.isdir(category_dir):
+                if await AsyncPath(category_dir).is_dir():
                     available_labels.add(category_name)
 
         if not available_labels:

frigate/api/media.py

@@ -15,6 +15,8 @@ from urllib.parse import unquote
 import cv2
 import numpy as np
 import pytz
+from anyio import Path as AsyncPath
+from anyio import open_file as aopen
 from fastapi import APIRouter, Depends, Path, Query, Request, Response
 from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
 from pathvalidate import sanitize_filename
@@ -497,18 +499,18 @@ async def recording_clip(
 
     file_name = sanitize_filename(f"playlist_{camera_name}_{start_ts}-{end_ts}.txt")
     file_path = os.path.join(CACHE_DIR, file_name)
-    with open(file_path, "w") as file:
+    async with await aopen(file_path, "w") as file:
         clip: Recordings
         for clip in recordings:
-            file.write(f"file '{clip.path}'\n")
+            await file.write(f"file '{clip.path}'\n")
 
             # if this is the starting clip, add an inpoint
             if clip.start_time < start_ts:
-                file.write(f"inpoint {int(start_ts - clip.start_time)}\n")
+                await file.write(f"inpoint {int(start_ts - clip.start_time)}\n")
 
             # if this is the ending clip, add an outpoint
             if clip.end_time > end_ts:
-                file.write(f"outpoint {int(end_ts - clip.start_time)}\n")
+                await file.write(f"outpoint {int(end_ts - clip.start_time)}\n")
 
     if len(file_name) > 1000:
         return JSONResponse(
@@ -1149,8 +1151,8 @@ async def event_snapshot_clean(request: Request, event_id: str, download: bool =
                 )
 
             if image_path.endswith(".webp"):
-                with open(image_path, "rb") as image_file:
+                async with await aopen(image_path, "rb") as image_file:
-                    webp_bytes = image_file.read()
+                    webp_bytes = await image_file.read()
             else:
                 image = load_event_snapshot_image(event, clean_only=True)[0]
                 if image is None:
@@ -1366,7 +1368,7 @@ async def preview_gif(
         # need to generate from existing images
         preview_dir = os.path.join(CACHE_DIR, "preview_frames")
 
-        if not os.path.isdir(preview_dir):
+        if not await AsyncPath(preview_dir).is_dir():
             return JSONResponse(
                 content={"success": False, "message": "Preview not found"},
                 status_code=404,
@@ -1555,7 +1557,7 @@ async def preview_mp4(
         # need to generate from existing images
         preview_dir = os.path.join(CACHE_DIR, "preview_frames")
 
-        if not os.path.isdir(preview_dir):
+        if not await AsyncPath(preview_dir).is_dir():
             return JSONResponse(
                 content={"success": False, "message": "Preview not found"},
                 status_code=404,
@@ -1633,7 +1635,7 @@ async def preview_mp4(
         "Content-Description": "File Transfer",
         "Cache-Control": f"private, max-age={_resolve_cache_age(max_cache_age)}",
         "Content-Type": "video/mp4",
-        "Content-Length": str(os.path.getsize(path)),
+        "Content-Length": str((await AsyncPath(path).stat()).st_size),
         # nginx: https://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_ignore_headers
         "X-Accel-Redirect": f"/cache/{file_name}",
     }
@@ -1707,10 +1709,10 @@ async def preview_thumbnail(request: Request, file_name: str):
     preview_dir = os.path.join(CACHE_DIR, "preview_frames")
 
     try:
-        with open(
+        async with await aopen(
             os.path.join(preview_dir, safe_file_name_current), "rb"
         ) as image_file:
-            jpg_bytes = image_file.read()
+            jpg_bytes = await image_file.read()
     except FileNotFoundError:
         return JSONResponse(
             content=({"success": False, "message": "Image file not found"}),

frigate/api/record.py

@@ -4,9 +4,9 @@ import datetime as dt
 import logging
 from datetime import datetime, timedelta
 from functools import reduce
-from pathlib import Path
 from urllib.parse import unquote
 
+from anyio import Path as AsyncPath
 from fastapi import APIRouter, Depends, Request
 from fastapi import Path as PathParam
 from fastapi.responses import JSONResponse
@@ -443,7 +443,7 @@ async def delete_recordings(
         recording_ids.append(recording["id"])
 
         try:
-            Path(recording["path"]).unlink(missing_ok=True)
+            await AsyncPath(recording["path"]).unlink(missing_ok=True)
             deleted_count += 1
         except Exception as e:
             logger.error(f"Failed to delete recording file {recording['path']}: {e}")

frigate/record/maintainer.py

@@ -15,6 +15,7 @@ from typing import Any
 
 import numpy as np
 import psutil
+from anyio import Path as AsyncPath
 
 from frigate.comms.detections_updater import DetectionSubscriber, DetectionTypeEnum
 from frigate.comms.inter_process import InterProcessRequestor
@@ -105,11 +106,11 @@ class RecordingMaintainer(threading.Thread):
 
     async def move_files(self) -> None:
         cache_files = [
-            d
+            path.name
-            for d in os.listdir(CACHE_DIR)
+            async for path in AsyncPath(CACHE_DIR).iterdir()
-            if os.path.isfile(os.path.join(CACHE_DIR, d))
+            if await path.is_file()
-            and d.endswith(".mp4")
+            and path.suffix == ".mp4"
-            and not d.startswith("preview_")
+            and not path.name.startswith("preview_")
         ]
 
         # publish newest cached segment per camera (including in use files)
@@ -229,7 +230,7 @@ class RecordingMaintainer(threading.Thread):
                 to_remove = grouped_recordings[camera][:-keep_count]
                 for rec in to_remove:
                     cache_path = rec["cache_path"]
-                    Path(cache_path).unlink(missing_ok=True)
+                    await AsyncPath(cache_path).unlink(missing_ok=True)
                     self.end_time_cache.pop(cache_path, None)
                 grouped_recordings[camera] = grouped_recordings[camera][-keep_count:]
 
@@ -244,7 +245,7 @@ class RecordingMaintainer(threading.Thread):
                 to_remove = grouped_recordings[camera][:-keep_count]
                 for rec in to_remove:
                     cache_path = rec["cache_path"]
-                    Path(cache_path).unlink(missing_ok=True)
+                    await AsyncPath(cache_path).unlink(missing_ok=True)
                     self.end_time_cache.pop(cache_path, None)
                 grouped_recordings[camera] = grouped_recordings[camera][-keep_count:]
 
@@ -634,7 +635,7 @@ class RecordingMaintainer(threading.Thread):
         file_path = os.path.join(directory, file_name)
 
         try:
-            if not os.path.exists(file_path):
+            if not await AsyncPath(file_path).exists():
                 start_frame = datetime.datetime.now().timestamp()
 
                 # add faststart to kept segments to improve metadata reading
@@ -670,7 +671,9 @@ class RecordingMaintainer(threading.Thread):
                     # get the segment size of the cache file
                     # file without faststart is same size
                     segment_size = round(
-                        float(os.path.getsize(cache_path)) / pow(2, 20), 2
+                        float((await AsyncPath(cache_path).stat()).st_size)
+                        / pow(2, 20),
+                        2,
                     )
                 except OSError:
                     segment_size = 0
@@ -698,7 +701,7 @@ class RecordingMaintainer(threading.Thread):
                 }
         except Exception as e:
             logger.error(f"Unable to store recording segment {cache_path}")
-            Path(cache_path).unlink(missing_ok=True)
+            await AsyncPath(cache_path).unlink(missing_ok=True)
             logger.error(e)
 
         # clear end_time cache

frigate/test/test_maintainer.py

@@ -1,7 +1,7 @@
 import datetime
 import sys
 import unittest
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 # Mock complex imports before importing maintainer, saving originals so we can
 # restore them after import and avoid polluting sys.modules for other tests.
@@ -42,38 +42,51 @@ class TestMaintainer(unittest.IsolatedAsyncioTestCase):
         # One bad file, one good file
         files = ["bad_filename.mp4", "camera@20210101000000+0000.mp4"]
 
-        with patch("os.listdir", return_value=files):
+        mock_paths = []
-            with patch("os.path.isfile", return_value=True):
+        for filename in files:
-                with patch(
+            path = MagicMock()
-                    "frigate.record.maintainer.psutil.process_iter", return_value=[]
+            path.name = filename
-                ):
+            path.suffix = ".mp4"
-                    with patch("frigate.record.maintainer.logger.warning") as warn:
+            path.is_file = AsyncMock(return_value=True)
-                        # Mock validate_and_move_segment to avoid further logic
+            mock_paths.append(path)
-                        maintainer.validate_and_move_segment = MagicMock()
 
-                        try:
+        async def mock_iterdir():
-                            await maintainer.move_files()
+            for path in mock_paths:
-                        except ValueError as e:
+                yield path
-                            if "not enough values to unpack" in str(e):
-                                self.fail("move_files() crashed on bad filename!")
-                            raise e
-                        except Exception:
-                            # Ignore other errors (like DB connection) as we only care about the unpack crash
-                            pass
 
-                        # The bad filename is encountered in multiple loops, but should only warn once.
+        with patch("frigate.record.maintainer.AsyncPath") as mock_async_path:
-                        matching = [
+            mock_async_path.return_value.iterdir = mock_iterdir
-                            c
+
-                            for c in warn.call_args_list
+            with patch(
-                            if c.args
+                "frigate.record.maintainer.psutil.process_iter", return_value=[]
-                            and isinstance(c.args[0], str)
+            ):
-                            and "Skipping unexpected files in cache" in c.args[0]
+                with patch("frigate.record.maintainer.logger.warning") as warn:
-                        ]
+                    # Mock validate_and_move_segment to avoid further logic
-                        self.assertEqual(
+                    maintainer.validate_and_move_segment = MagicMock()
-                            1,
+
-                            len(matching),
+                    try:
-                            f"Expected a single warning for unexpected files, got {len(matching)}",
+                        await maintainer.move_files()
-                        )
+                    except ValueError as e:
+                        if "not enough values to unpack" in str(e):
+                            self.fail("move_files() crashed on bad filename!")
+                        raise e
+                    except Exception:
+                        # Ignore other errors (like DB connection) as we only care about the unpack crash
+                        pass
+
+                    # The bad filename is encountered in multiple loops, but should only warn once.
+                    matching = [
+                        c
+                        for c in warn.call_args_list
+                        if c.args
+                        and isinstance(c.args[0], str)
+                        and "Skipping unexpected files in cache" in c.args[0]
+                    ]
+                    self.assertEqual(
+                        1,
+                        len(matching),
+                        f"Expected a single warning for unexpected files, got {len(matching)}",
+                    )
 
     async def test_drops_quiet_segment_when_only_motion_retention(self):
         # Regression: when motion retention is enabled but a segment has no

pyproject.toml

@@ -2,5 +2,20 @@
 target-version = "py311"
 
 [tool.ruff.lint]
-ignore = ["E501","E711","E712","UP031","UP032","UP042","G004"]
+ignore = [
-extend-select = ["I", "UP", "G", "ASYNC210", "B904"]
+  "ASYNC109", # Async function definition with a timeout parameter
+  "E501",     # line-too-long
+  "E711",     # none-comparison
+  "E712",     # true-false-comparison
+  "UP031",    # printf-string-formatting
+  "UP032",    # f-string
+  "UP042",    # replace-str-enum
+  "G004",     # logging-f-string
+]
+extend-select = [
+  "ASYNC", # https://docs.astral.sh/ruff/rules/#flake8-async-async
+  "B904",  # https://docs.astral.sh/ruff/rules/raise-without-from-inside-except/
+  "G",     # https://docs.astral.sh/ruff/rules/#flake8-logging-format-g
+  "I",     # https://docs.astral.sh/ruff/rules/#isort-i
+  "UP",    # https://docs.astral.sh/ruff/rules/#pyupgrade-up
+]