20108598472

Committed 10 Dec 2025 06:06PM UTC coverage: 92.575% (-0.2%) from 92.793%

Build # 20108598472

Build Type

push

github

Committed by

agronholm

Commit Message

Removed unconditional import of sniffio

Sniffio is no longer an implicit dependency, as of AnyIO 4.12.0.

Fixes #1093.

Coverage Stats

8 of 15 new or added lines in 2 files covered. (53.33%)

7 existing lines in 2 files now uncovered.

3491 of 3771 relevant lines covered (92.57%)

7.67 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.71

/src/apscheduler/_utils.py

"""This module contains several handy functions primarily meant for internal use."""

from __future__ import annotations

import asyncio
from datetime import datetime, tzinfo
from typing import TYPE_CHECKING, Any, NoReturn, TypeVar
from zoneinfo import ZoneInfo

from ._exceptions import DeserializationError
from .abc import Trigger

try:
    import sniffio
except ImportError:
    sniffio = None

if TYPE_CHECKING:
    from ._structures import MetadataType

T = TypeVar("T")


class UnsetValue:
    """The type of :data:`unset`."""

    __slots__ = ()

    def __new__(cls) -> UnsetValue:
        try:
            return unset
        except NameError:
            return super().__new__(cls)

    def __getstate__(self) -> NoReturn:
        raise RuntimeError("Internal error: attempted to serialize an unset value")

    def __repr__(self) -> str:
        return "<unset>"


unset = UnsetValue()


def timezone_repr(timezone: tzinfo) -> str:
    if isinstance(timezone, ZoneInfo):
        return timezone.key
    else:
        return repr(timezone)


def absolute_datetime_diff(dateval1: datetime, dateval2: datetime) -> float:
    return dateval1.timestamp() - dateval2.timestamp()


def qualified_name(cls: type) -> str:
    module = getattr(cls, "__module__", None)
    if module is None or module == "builtins":
        return cls.__qualname__
    else:
        return f"{module}.{cls.__qualname__}"


def require_state_version(
    trigger: Trigger, state: dict[str, Any], max_version: int
) -> None:
    try:
        if state["version"] > max_version:
            raise DeserializationError(
                f"{trigger.__class__.__name__} received a serialized state with "
                f"version {state['version']}, but it only supports up to version "
                f"{max_version}. This can happen when an older version of APScheduler "
                f"is being used with a data store that was previously used with a "
                f"newer APScheduler version."
            )
    except KeyError as exc:
        raise DeserializationError(
            'Missing "version" key in the serialized state'
        ) from exc


def merge_metadata(
    base_metadata: MetadataType, *overlays: MetadataType | UnsetValue
) -> MetadataType:
    new_metadata = base_metadata.copy()
    for metadata in overlays:
        if isinstance(metadata, UnsetValue):
            continue

        new_metadata.update(metadata)

    return new_metadata


def create_repr(instance: object, *attrnames: str, **kwargs) -> str:
    kv_pairs: list[tuple[str, object]] = []
    for attrname in attrnames:
        value = getattr(instance, attrname)
        if value is not unset and value is not None:
            kv_pairs.append((attrname, value))

    for key, value in kwargs.items():
        if value is not unset and value is not None:
            kv_pairs.append((key, value))

    rendered_attrs = ", ".join(f"{key}={value!r}" for key, value in kv_pairs)
    return f"{instance.__class__.__name__}({rendered_attrs})"


def time_exists(dt: datetime) -> bool:
    """
    Determine whether a datetime exists in its time zone.

    :return: ``False`` if the given datetime falls within a gap created by a
        forward daylight savings shift, otherwise ``True``

    """
    return dt == datetime.fromtimestamp(dt.timestamp(), dt.tzinfo)


def current_async_library() -> str:
    """Return the name of the currently used async library."""
    if sniffio is not None:
        return sniffio.current_async_library() or "(unknown)"

    try:
        asyncio.get_running_loop()
        return "asyncio"
    except RuntimeError:
        return "(unknown)"

1	"""This module contains several handy functions primarily meant for internal use."""
2
3	from __future__ import annotations	9✔
4
5	import asyncio	9✔
6	from datetime import datetime, tzinfo	9✔
7	from typing import TYPE_CHECKING, Any, NoReturn, TypeVar	9✔
8	from zoneinfo import ZoneInfo	9✔
9
10	from ._exceptions import DeserializationError	9✔
11	from .abc import Trigger	9✔
12
13	try:	9✔
14	import sniffio	9✔
NEW 15	except ImportError:	×
NEW 16	sniffio = None	×
17
18	if TYPE_CHECKING:
19	from ._structures import MetadataType
20
21	T = TypeVar("T")	9✔
22
23
24	class UnsetValue:	9✔
25	"""The type of :data:`unset`."""
26
27	__slots__ = ()	9✔
28
29	def __new__(cls) -> UnsetValue:	9✔
30	try:	9✔
31	return unset	9✔
32	except NameError:	9✔
33	return super().__new__(cls)	9✔
34
35	def __getstate__(self) -> NoReturn:	9✔
36	raise RuntimeError("Internal error: attempted to serialize an unset value")	×
37
38	def __repr__(self) -> str:	9✔
39	return "<unset>"	×
40
41
42	unset = UnsetValue()	9✔
43
44
45	def timezone_repr(timezone: tzinfo) -> str:	9✔
46	if isinstance(timezone, ZoneInfo):	9✔
47	return timezone.key	9✔
48	else:
49	return repr(timezone)	×
50
51
52	def absolute_datetime_diff(dateval1: datetime, dateval2: datetime) -> float:	9✔
53	return dateval1.timestamp() - dateval2.timestamp()	×
54
55
56	def qualified_name(cls: type) -> str:	9✔
57	module = getattr(cls, "__module__", None)	9✔
58	if module is None or module == "builtins":	9✔
59	return cls.__qualname__	9✔
60	else:
61	return f"{module}.{cls.__qualname__}"	9✔
62
63
64	def require_state_version(	9✔
65	trigger: Trigger, state: dict[str, Any], max_version: int
66	) -> None:
67	try:	9✔
68	if state["version"] > max_version:	9✔
69	raise DeserializationError(	×
70	f"{trigger.__class__.__name__} received a serialized state with "
71	f"version {state['version']}, but it only supports up to version "
72	f"{max_version}. This can happen when an older version of APScheduler "
73	f"is being used with a data store that was previously used with a "
74	f"newer APScheduler version."
75	)
76	except KeyError as exc:	×
77	raise DeserializationError(	×
78	'Missing "version" key in the serialized state'
79	) from exc
80
81
82	def merge_metadata(	9✔
83	base_metadata: MetadataType, *overlays: MetadataType \| UnsetValue
84	) -> MetadataType:
85	new_metadata = base_metadata.copy()	9✔
86	for metadata in overlays:	9✔
87	if isinstance(metadata, UnsetValue):	9✔
88	continue	9✔
89
90	new_metadata.update(metadata)	9✔
91
92	return new_metadata	9✔
93
94
95	def create_repr(instance: object, attrnames: str, *kwargs) -> str:	9✔
96	kv_pairs: list[tuple[str, object]] = []	9✔
97	for attrname in attrnames:	9✔
98	value = getattr(instance, attrname)	9✔
99	if value is not unset and value is not None:	9✔
100	kv_pairs.append((attrname, value))	9✔
101
102	for key, value in kwargs.items():	9✔
103	if value is not unset and value is not None:	9✔
104	kv_pairs.append((key, value))	9✔
105
106	rendered_attrs = ", ".join(f"{key}={value!r}" for key, value in kv_pairs)	9✔
107	return f"{instance.__class__.__name__}({rendered_attrs})"	9✔
108
109
110	def time_exists(dt: datetime) -> bool:	9✔
111	"""
112	Determine whether a datetime exists in its time zone.
113
114	:return: ``False`` if the given datetime falls within a gap created by a
115	forward daylight savings shift, otherwise ``True``
116
117	"""
118	return dt == datetime.fromtimestamp(dt.timestamp(), dt.tzinfo)	9✔
119
120
121	def current_async_library() -> str:	9✔
122	"""Return the name of the currently used async library."""
123	if sniffio is not None:	9✔
124	return sniffio.current_async_library() or "(unknown)"	9✔
125
NEW 126	try:	×
NEW 127	asyncio.get_running_loop()	×
NEW 128	return "asyncio"	×
NEW 129	except RuntimeError:	×
NEW 130	return "(unknown)"	×

agronholm / apscheduler / 20108598472

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous