Fix various warnings and deprecations (#23641)

* Migrate to fastapi lifespan for startup events

The `on_event` style decorator is deprecated

https://fastapi.tiangolo.com/advanced/events/

* Migrate to pydantic TypeAdapter

https://pydantic.dev/docs/validation/2.12/get-started/migration/#introduction-of-typeadapter

* Replace pytz with native tz-aware objects and zoneinfo library

Supported from Python 3.9, see https://peps.python.org/pep-0615/. Also
https://blog.ganssle.io/articles/2019/11/utcnow.html.

Also provide tzdata as new dependency to supplement systems without local
tzdata information.

* Provide lookup for normalized timezone names

This provides a helper to look up the correct timezone name independent
of the casing of the given timezone name.

* Test timezone naming normalization and error handling
This commit is contained in:
Martin Weinelt
2026-10-11 17:49:33 -05:00
committed by GitHub
parent 186a730790
commit 135a95d8f2
16 changed files with 115 additions and 68 deletions
+1 -1
View File
@@ -25,7 +25,7 @@ peewee_migrate == 2.3.*
psutil == 7.2.* psutil == 7.2.*
pydantic == 2.13.* pydantic == 2.13.*
git+https://github.com/fbcotter/py3nvml#egg=py3nvml git+https://github.com/fbcotter/py3nvml#egg=py3nvml
pytz == 2026.* tzdata == 2026.5
pyzmq == 27.2.* pyzmq == 27.2.*
ruamel.yaml == 0.19.* ruamel.yaml == 0.19.*
tzlocal == 5.4.4 tzlocal == 5.4.4
+7 -7
View File
@@ -2131,7 +2131,7 @@ paths:
required: false required: false
schema: schema:
type: string type: string
default: utc default: UTC
title: Timezone title: Timezone
responses: responses:
'200': '200':
@@ -3266,7 +3266,7 @@ paths:
anyOf: anyOf:
- type: string - type: string
- type: 'null' - type: 'null'
default: utc default: UTC
title: Timezone title: Timezone
responses: responses:
'200': '200':
@@ -4666,7 +4666,7 @@ paths:
anyOf: anyOf:
- type: string - type: string
- type: 'null' - type: 'null'
default: utc default: UTC
title: Timezone title: Timezone
responses: responses:
'200': '200':
@@ -4928,7 +4928,7 @@ paths:
anyOf: anyOf:
- type: string - type: string
- type: 'null' - type: 'null'
default: utc default: UTC
title: Timezone title: Timezone
- name: min_score - name: min_score
in: query in: query
@@ -5008,7 +5008,7 @@ paths:
anyOf: anyOf:
- type: string - type: string
- type: 'null' - type: 'null'
default: utc default: UTC
title: Timezone title: Timezone
- name: has_clip - name: has_clip
in: query in: query
@@ -7367,7 +7367,7 @@ paths:
required: false required: false
schema: schema:
type: string type: string
default: utc default: UTC
title: Timezone title: Timezone
- name: cameras - name: cameras
in: query in: query
@@ -7417,7 +7417,7 @@ paths:
required: false required: false
schema: schema:
type: string type: string
default: utc default: UTC
title: Timezone title: Timezone
responses: responses:
'200': '200':
@@ -7,4 +7,4 @@ class AppTimelineHourlyQueryParameters(BaseModel):
after: float | None = None after: float | None = None
before: float | None = None before: float | None = None
limit: int | None = 200 limit: int | None = 200
timezone: str | None = "utc" timezone: str | None = "UTC"
@@ -40,7 +40,7 @@ class EventsQueryParams(BaseModel):
max_length: float | None = None max_length: float | None = None
event_id: str | None = None event_id: str | None = None
sort: str | None = None sort: str | None = None
timezone: str | None = "utc" timezone: str | None = "UTC"
class EventsSearchQueryParams(BaseModel): class EventsSearchQueryParams(BaseModel):
@@ -68,7 +68,7 @@ class EventsSearchQueryParams(BaseModel):
has_clip: bool | None = None has_clip: bool | None = None
has_snapshot: bool | None = None has_snapshot: bool | None = None
is_submitted: bool | None = None is_submitted: bool | None = None
timezone: str | None = "utc" timezone: str | None = "UTC"
min_score: float | None = None min_score: float | None = None
max_score: float | None = None max_score: float | None = None
min_speed: float | None = None min_speed: float | None = None
@@ -78,6 +78,6 @@ class EventsSearchQueryParams(BaseModel):
class EventsSummaryQueryParams(BaseModel): class EventsSummaryQueryParams(BaseModel):
timezone: str | None = "utc" timezone: str | None = "UTC"
has_clip: int | None = None has_clip: int | None = None
has_snapshot: int | None = None has_snapshot: int | None = None
@@ -3,7 +3,7 @@ from pydantic.json_schema import SkipJsonSchema
class MediaRecordingsSummaryQueryParams(BaseModel): class MediaRecordingsSummaryQueryParams(BaseModel):
timezone: str = "utc" timezone: str = "UTC"
cameras: str | None = "all" cameras: str | None = "all"
@@ -19,7 +19,7 @@ class ReviewSummaryQueryParams(BaseModel):
cameras: str = "all" cameras: str = "all"
labels: str = "all" labels: str = "all"
zones: str = "all" zones: str = "all"
timezone: str = "utc" timezone: str = "UTC"
class ReviewActivityMotionQueryParams(BaseModel): class ReviewActivityMotionQueryParams(BaseModel):
+12 -9
View File
@@ -1,6 +1,7 @@
import asyncio import asyncio
import logging import logging
import re import re
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI, Request from fastapi import Depends, FastAPI, Request
from fastapi.responses import JSONResponse from fastapi.responses import JSONResponse
@@ -81,9 +82,20 @@ def create_fastapi_app(
config_holder: ConfigHolder | None = None, config_holder: ConfigHolder | None = None,
notice_registry: NoticeRegistry | None = None, notice_registry: NoticeRegistry | None = None,
): ):
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("FastAPI started")
asyncio.create_task(
debug_replay_auto_stop_watchdog(
replay_manager, frigate_config, config_publisher
)
)
yield
logger.info("Starting FastAPI app") logger.info("Starting FastAPI app")
app = FastAPI( app = FastAPI(
debug=False, debug=False,
lifespan=lifespan,
swagger_ui_parameters={"apisSorter": "alpha", "operationsSorter": "alpha"}, swagger_ui_parameters={"apisSorter": "alpha", "operationsSorter": "alpha"},
dependencies=[Depends(require_admin_by_default())] dependencies=[Depends(require_admin_by_default())]
if enforce_default_admin if enforce_default_admin
@@ -119,15 +131,6 @@ def create_fastapi_app(
database.close() database.close()
return response return response
@app.on_event("startup")
async def startup():
logger.info("FastAPI started")
asyncio.create_task(
debug_replay_auto_stop_watchdog(
replay_manager, frigate_config, config_publisher
)
)
# Rate limiter (used for login endpoint) # Rate limiter (used for login endpoint)
if frigate_config.auth.failed_login_rate_limit is None: if frigate_config.auth.failed_login_rate_limit is None:
limiter.enabled = False limiter.enabled = False
+5 -2
View File
@@ -14,10 +14,10 @@ from enum import Enum
from pathlib import Path as FilePath from pathlib import Path as FilePath
from typing import IO, Any from typing import IO, Any
from urllib.parse import unquote from urllib.parse import unquote
from zoneinfo import ZoneInfo
import cv2 import cv2
import numpy as np import numpy as np
import pytz
from fastapi import APIRouter, Depends, Path, Query, Request, Response from fastapi import APIRouter, Depends, Path, Query, Request, Response
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from pathvalidate import sanitize_filename from pathvalidate import sanitize_filename
@@ -64,6 +64,7 @@ from frigate.util.recording_coverage import (
plan_clip, plan_clip,
resolve_coverage, resolve_coverage,
) )
from frigate.util.time import get_normalized_tz_name
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -837,7 +838,9 @@ async def vod_hour(
parts = year_month.split("-") parts = year_month.split("-")
start_date = ( start_date = (
datetime(int(parts[0]), int(parts[1]), day, hour, tzinfo=UTC) datetime(int(parts[0]), int(parts[1]), day, hour, tzinfo=UTC)
- datetime.now(pytz.timezone(tz_name.replace(",", "/"))).utcoffset() - datetime.now(
ZoneInfo(get_normalized_tz_name(tz_name.replace(",", "/")))
).utcoffset()
) )
end_date = start_date + timedelta(hours=1) - timedelta(milliseconds=1) end_date = start_date + timedelta(hours=1) - timedelta(milliseconds=1)
start_ts = start_date.timestamp() start_ts = start_date.timestamp()
+5 -2
View File
@@ -5,8 +5,8 @@ import logging
import os import os
import threading import threading
from datetime import UTC, datetime, timedelta from datetime import UTC, datetime, timedelta
from zoneinfo import ZoneInfo
import pytz
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse from fastapi.responses import JSONResponse
@@ -22,6 +22,7 @@ from frigate.api.defs.response.preview_response import (
from frigate.api.defs.tags import Tags from frigate.api.defs.tags import Tags
from frigate.const import BASE_DIR, CACHE_DIR, PREVIEW_FRAME_TYPE from frigate.const import BASE_DIR, CACHE_DIR, PREVIEW_FRAME_TYPE
from frigate.models import Previews from frigate.models import Previews
from frigate.util.time import get_normalized_tz_name
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -126,7 +127,9 @@ def preview_hour(
parts = year_month.split("-") parts = year_month.split("-")
start_date = ( start_date = (
datetime(int(parts[0]), int(parts[1]), int(day), int(hour), tzinfo=UTC) datetime(int(parts[0]), int(parts[1]), int(day), int(hour), tzinfo=UTC)
- datetime.now(pytz.timezone(tz_name.replace(",", "/"))).utcoffset() - datetime.now(
ZoneInfo(get_normalized_tz_name(tz_name.replace(",", "/")))
).utcoffset()
) )
end_date = start_date + timedelta(hours=1) - timedelta(milliseconds=1) end_date = start_date + timedelta(hours=1) - timedelta(milliseconds=1)
start_ts = start_date.timestamp() start_ts = start_date.timestamp()
+1 -1
View File
@@ -175,7 +175,7 @@ async def all_recordings_summary(
@router.get( @router.get(
"/{camera_name}/recordings/summary", dependencies=[Depends(require_camera_access)] "/{camera_name}/recordings/summary", dependencies=[Depends(require_camera_access)]
) )
async def recordings_summary(camera_name: str, timezone: str = "utc"): async def recordings_summary(camera_name: str, timezone: str = "UTC"):
"""Returns hourly summary for recordings of given camera""" """Returns hourly summary for recordings of given camera"""
time_range_query = ( time_range_query = (
+6 -6
View File
@@ -15,8 +15,8 @@ from dataclasses import dataclass
from enum import Enum from enum import Enum
from pathlib import Path from pathlib import Path
from typing import Any, cast from typing import Any, cast
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import pytz
from pathvalidate import sanitize_filename from pathvalidate import sanitize_filename
from peewee import DoesNotExist from peewee import DoesNotExist
@@ -49,7 +49,7 @@ from frigate.util.recording_coverage import (
stream_media_summary, stream_media_summary,
) )
from frigate.util.services import get_video_properties from frigate.util.services import get_video_properties
from frigate.util.time import is_current_hour from frigate.util.time import get_normalized_tz_name, is_current_hour
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -874,8 +874,8 @@ class RecordingExporter(threading.Thread):
tz_name = self.config.ui.timezone tz_name = self.config.ui.timezone
if tz_name: if tz_name:
try: try:
tz = pytz.timezone(tz_name) tz = ZoneInfo(get_normalized_tz_name(tz_name))
except pytz.UnknownTimeZoneError: except (ValueError, ZoneInfoNotFoundError):
tz = None tz = None
if tz is not None: if tz is not None:
return datetime.datetime.fromtimestamp(timestamp, tz=tz).strftime( return datetime.datetime.fromtimestamp(timestamp, tz=tz).strftime(
@@ -1038,8 +1038,8 @@ class RecordingExporter(threading.Thread):
tz: datetime.tzinfo | None = None tz: datetime.tzinfo | None = None
if tz_name: if tz_name:
try: try:
tz = pytz.timezone(tz_name) tz = ZoneInfo(get_normalized_tz_name(tz_name))
except pytz.UnknownTimeZoneError: except (ValueError, ZoneInfoNotFoundError):
tz = None tz = None
if tz is None: if tz is None:
tz = datetime.UTC tz = datetime.UTC
+18 -21
View File
@@ -3,8 +3,8 @@
from dataclasses import dataclass from dataclasses import dataclass
from datetime import UTC, datetime, timedelta from datetime import UTC, datetime, timedelta
from unittest.mock import patch from unittest.mock import patch
from zoneinfo import ZoneInfo
import pytz
from fastapi import Request from fastapi import Request
from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user
@@ -128,16 +128,16 @@ class TestHttpMedia(BaseTestHttp):
In 2024, DST in America/New_York transitions on March 10, 2024 at 2:00 AM In 2024, DST in America/New_York transitions on March 10, 2024 at 2:00 AM
Clocks spring forward from 2:00 AM to 3:00 AM (EST to EDT) Clocks spring forward from 2:00 AM to 3:00 AM (EST to EDT)
""" """
tz = pytz.timezone("America/New_York") tz = ZoneInfo("America/New_York")
# March 9, 2024 at 12:00 PM EST (before DST) # March 9, 2024 at 12:00 PM EST (before DST)
march_9_noon = tz.localize(datetime(2024, 3, 9, 12, 0, 0)).timestamp() march_9_noon = datetime(2024, 3, 9, 12, 0, 0, tzinfo=tz).timestamp()
# March 10, 2024 at 12:00 PM EDT (after DST transition) # March 10, 2024 at 12:00 PM EDT (after DST transition)
march_10_noon = tz.localize(datetime(2024, 3, 10, 12, 0, 0)).timestamp() march_10_noon = datetime(2024, 3, 10, 12, 0, 0, tzinfo=tz).timestamp()
# March 11, 2024 at 12:00 PM EDT (after DST) # March 11, 2024 at 12:00 PM EDT (after DST)
march_11_noon = tz.localize(datetime(2024, 3, 11, 12, 0, 0)).timestamp() march_11_noon = datetime(2024, 3, 11, 12, 0, 0, tzinfo=tz).timestamp()
with AuthTestClient(self.app) as client: with AuthTestClient(self.app) as client:
# Insert recordings for each day # Insert recordings for each day
@@ -201,19 +201,16 @@ class TestHttpMedia(BaseTestHttp):
In 2024, DST in America/New_York transitions on November 3, 2024 at 2:00 AM In 2024, DST in America/New_York transitions on November 3, 2024 at 2:00 AM
Clocks fall back from 2:00 AM to 1:00 AM (EDT to EST) Clocks fall back from 2:00 AM to 1:00 AM (EDT to EST)
""" """
tz = pytz.timezone("America/New_York") tz = ZoneInfo("America/New_York")
# November 2, 2024 at 12:00 PM EDT (before DST transition) # November 2, 2024 at 12:00 PM EDT (before DST transition)
nov_2_noon = tz.localize(datetime(2024, 11, 2, 12, 0, 0)).timestamp() nov_2_noon = datetime(2024, 11, 2, 12, 0, 0, tzinfo=tz).timestamp()
# November 3, 2024 at 12:00 PM EST (after DST transition) # November 3, 2024 at 12:00 PM EST (after DST transition)
# Need to specify is_dst=False to get the time after fall back nov_3_noon = datetime(2024, 11, 3, 12, 0, 0, tzinfo=tz).timestamp()
nov_3_noon = tz.localize(
datetime(2024, 11, 3, 12, 0, 0), is_dst=False
).timestamp()
# November 4, 2024 at 12:00 PM EST (after DST) # November 4, 2024 at 12:00 PM EST (after DST)
nov_4_noon = tz.localize(datetime(2024, 11, 4, 12, 0, 0)).timestamp() nov_4_noon = datetime(2024, 11, 4, 12, 0, 0, tzinfo=tz).timestamp()
with AuthTestClient(self.app) as client: with AuthTestClient(self.app) as client:
# Insert recordings for each day # Insert recordings for each day
@@ -274,13 +271,13 @@ class TestHttpMedia(BaseTestHttp):
""" """
Test recordings summary with multiple cameras across DST boundary. Test recordings summary with multiple cameras across DST boundary.
""" """
tz = pytz.timezone("America/New_York") tz = ZoneInfo("America/New_York")
# March 9, 2024 at 10:00 AM EST (before DST) # March 9, 2024 at 10:00 AM EST (before DST)
march_9_morning = tz.localize(datetime(2024, 3, 9, 10, 0, 0)).timestamp() march_9_morning = datetime(2024, 3, 9, 10, 0, 0, tzinfo=tz).timestamp()
# March 10, 2024 at 3:00 PM EDT (after DST transition) # March 10, 2024 at 3:00 PM EDT (after DST transition)
march_10_afternoon = tz.localize(datetime(2024, 3, 10, 15, 0, 0)).timestamp() march_10_afternoon = datetime(2024, 3, 10, 15, 0, 0, tzinfo=tz).timestamp()
with AuthTestClient(self.app) as client: with AuthTestClient(self.app) as client:
# Override allowed cameras for this test to include both # Override allowed cameras for this test to include both
@@ -343,15 +340,15 @@ class TestHttpMedia(BaseTestHttp):
""" """
Test recordings that span the exact DST transition time. Test recordings that span the exact DST transition time.
""" """
tz = pytz.timezone("America/New_York") tz = ZoneInfo("America/New_York")
# March 10, 2024 at 1:00 AM EST (1 hour before DST transition) # March 10, 2024 at 1:00 AM EST (1 hour before DST transition)
# At 2:00 AM, clocks jump to 3:00 AM # At 2:00 AM, clocks jump to 3:00 AM
before_transition = tz.localize(datetime(2024, 3, 10, 1, 0, 0)).timestamp() before_transition = datetime(2024, 3, 10, 1, 0, 0, tzinfo=tz).timestamp()
# Recording that spans the transition (1:00 AM to 3:30 AM EDT) # Recording that spans the transition (1:00 AM to 3:30 AM EDT)
# This is 1.5 hours of actual time but spans the "missing" hour # This is 1.5 hours of actual time but spans the "missing" hour
after_transition = tz.localize(datetime(2024, 3, 10, 3, 30, 0)).timestamp() after_transition = datetime(2024, 3, 10, 3, 30, 0, tzinfo=tz).timestamp()
with AuthTestClient(self.app) as client: with AuthTestClient(self.app) as client:
Recordings.insert( Recordings.insert(
@@ -411,7 +408,7 @@ class TestHttpMedia(BaseTestHttp):
# Test with UTC timezone # Test with UTC timezone
response = client.get( response = client.get(
"/recordings/summary", params={"timezone": "utc", "cameras": "all"} "/recordings/summary", params={"timezone": "UTC", "cameras": "all"}
) )
assert response.status_code == 200 assert response.status_code == 200
@@ -442,8 +439,8 @@ class TestHttpMedia(BaseTestHttp):
""" """
Test recordings summary filtered to a single camera. Test recordings summary filtered to a single camera.
""" """
tz = pytz.timezone("America/New_York") tz = ZoneInfo("America/New_York")
march_10_noon = tz.localize(datetime(2024, 3, 10, 12, 0, 0)).timestamp() march_10_noon = datetime(2024, 3, 10, 12, 0, 0, tzinfo=tz).timestamp()
with AuthTestClient(self.app) as client: with AuthTestClient(self.app) as client:
# Insert recordings for both cameras # Insert recordings for both cameras
+1 -1
View File
@@ -342,7 +342,7 @@ class TestHttpReview(BaseTestHttp):
"cameras": "front_door", "cameras": "front_door",
"labels": "all", "labels": "all",
"zones": "all", "zones": "all",
"timezone": "utc", "timezone": "UTC",
} }
response = client.get("/review/summary", params=params) response = client.get("/review/summary", params=params)
assert response.status_code == 200 assert response.status_code == 200
+12 -6
View File
@@ -3,7 +3,7 @@ from unittest.mock import MagicMock, Mock, patch
import numpy as np import numpy as np
import zmq import zmq
from pydantic import parse_obj_as from pydantic import TypeAdapter
import frigate.detectors as detectors import frigate.detectors as detectors
import frigate.object_detection.base import frigate.object_detection.base
@@ -20,8 +20,8 @@ class TestLocalObjectDetector(unittest.TestCase):
"frigate.detectors.api_types", "frigate.detectors.api_types",
{det_type: Mock() for det_type in DetectorTypeEnum}, {det_type: Mock() for det_type in DetectorTypeEnum},
): ):
test_cfg = parse_obj_as( test_cfg = TypeAdapter(DetectorConfig).validate_python(
DetectorConfig, ({"type": det_type, "model": {}}) {"type": det_type, "model": {}}
) )
test_cfg.model.path = "/test/modelpath" test_cfg.model.path = "/test/modelpath"
test_obj = frigate.object_detection.base.LocalObjectDetector( test_obj = frigate.object_detection.base.LocalObjectDetector(
@@ -45,7 +45,9 @@ class TestLocalObjectDetector(unittest.TestCase):
TEST_DATA = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] TEST_DATA = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
TEST_DETECT_RESULT = np.ndarray([1, 2, 4, 8, 16, 32]) TEST_DETECT_RESULT = np.ndarray([1, 2, 4, 8, 16, 32])
test_obj_detect = frigate.object_detection.base.LocalObjectDetector( test_obj_detect = frigate.object_detection.base.LocalObjectDetector(
detector_config=parse_obj_as(DetectorConfig, {"type": "cpu", "model": {}}) detector_config=TypeAdapter(DetectorConfig).validate_python(
{"type": "cpu", "model": {}}
)
) )
mock_det_api = mock_cputfl.return_value mock_det_api = mock_cputfl.return_value
@@ -68,7 +70,9 @@ class TestLocalObjectDetector(unittest.TestCase):
TEST_DATA = np.zeros((1, 32, 32, 3), np.uint8) TEST_DATA = np.zeros((1, 32, 32, 3), np.uint8)
TEST_DETECT_RESULT = np.ndarray([1, 2, 4, 8, 16, 32]) TEST_DETECT_RESULT = np.ndarray([1, 2, 4, 8, 16, 32])
test_cfg = parse_obj_as(DetectorConfig, {"type": "cpu", "model": {}}) test_cfg = TypeAdapter(DetectorConfig).validate_python(
{"type": "cpu", "model": {}}
)
test_cfg.model.input_tensor = InputTensorEnum.nchw test_cfg.model.input_tensor = InputTensorEnum.nchw
test_obj_detect = frigate.object_detection.base.LocalObjectDetector( test_obj_detect = frigate.object_detection.base.LocalObjectDetector(
@@ -117,7 +121,9 @@ class TestLocalObjectDetector(unittest.TestCase):
4: "label-5", 4: "label-5",
} }
test_cfg = parse_obj_as(DetectorConfig, {"type": "cpu", "model": {}}) test_cfg = TypeAdapter(DetectorConfig).validate_python(
{"type": "cpu", "model": {}}
)
test_cfg.model = ModelConfig() test_cfg.model = ModelConfig()
test_obj_detect = frigate.object_detection.base.LocalObjectDetector( test_obj_detect = frigate.object_detection.base.LocalObjectDetector(
detector_config=test_cfg, detector_config=test_cfg,
+20
View File
@@ -0,0 +1,20 @@
import unittest
from frigate.util.time import get_normalized_tz_name
class TestGetNormalizedTzName(unittest.TestCase):
def test_valid(self):
cases = [
("utc", "UTC"),
("america/new_york", "America/New_York"),
("us/eastern", "US/Eastern"),
]
for tz_name, expected in cases:
with self.subTest(tz_name=tz_name):
self.assertEqual(get_normalized_tz_name(tz_name), expected)
def test_invalid(self):
with self.assertRaisesRegex(ValueError, r"Unknown timezone: foo/bar"):
get_normalized_tz_name("foo/bar")
+21 -6
View File
@@ -3,17 +3,32 @@
import datetime import datetime
import logging import logging
import math import math
from zoneinfo import ZoneInfoNotFoundError from typing import Final
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError, available_timezones
import pytz
from tzlocal import get_localzone from tzlocal import get_localzone
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
TIMEZONE_CASEFOLD_INDEX: Final = {
timezone.casefold(): timezone for timezone in available_timezones()
}
def get_normalized_tz_name(tz_name: str) -> str:
"""Return the canonical name for a case-insensitive IANA timezone."""
tz = TIMEZONE_CASEFOLD_INDEX.get(tz_name.casefold())
if tz:
return tz
raise ValueError(f"Unknown timezone: {tz_name}")
def get_tz_modifiers(tz_name: str) -> tuple[str, str, float]: def get_tz_modifiers(tz_name: str) -> tuple[str, str, float]:
seconds_offset = ( seconds_offset = (
datetime.datetime.now(pytz.timezone(tz_name)).utcoffset().total_seconds() datetime.datetime.now(ZoneInfo(get_normalized_tz_name(tz_name)))
.utcoffset()
.total_seconds()
) )
hours_offset = int(seconds_offset / 60 / 60) hours_offset = int(seconds_offset / 60 / 60)
minutes_offset = int(seconds_offset / 60 - hours_offset * 60) minutes_offset = int(seconds_offset / 60 - hours_offset * 60)
@@ -26,7 +41,7 @@ def get_tomorrow_at_time(hour: int) -> datetime.datetime:
"""Returns the datetime of the following day at 2am.""" """Returns the datetime of the following day at 2am."""
try: try:
tomorrow = datetime.datetime.now(get_localzone()) + datetime.timedelta(days=1) tomorrow = datetime.datetime.now(get_localzone()) + datetime.timedelta(days=1)
except ZoneInfoNotFoundError: except (ValueError, ZoneInfoNotFoundError):
tomorrow = datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=1) tomorrow = datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=1)
logger.warning( logger.warning(
"Using utc for maintenance due to missing or incorrect timezone set" "Using utc for maintenance due to missing or incorrect timezone set"
@@ -84,8 +99,8 @@ def get_dst_transitions(
continuous periods with the same UTC offset continuous periods with the same UTC offset
""" """
try: try:
tz = pytz.timezone(tz_name) tz = ZoneInfo(get_normalized_tz_name(tz_name))
except pytz.UnknownTimeZoneError: except (ValueError, ZoneInfoNotFoundError):
# If timezone is invalid, return single period with no offset # If timezone is invalid, return single period with no offset
return [(start_time, end_time, 0)] return [(start_time, end_time, 0)]