diff options
Diffstat (limited to 'src/apscheduler/structures.py')
| -rw-r--r-- | src/apscheduler/structures.py | 20 |
1 files changed, 10 insertions, 10 deletions
diff --git a/src/apscheduler/structures.py b/src/apscheduler/structures.py index eeab5ac..a25d411 100644 --- a/src/apscheduler/structures.py +++ b/src/apscheduler/structures.py @@ -1,7 +1,7 @@ from __future__ import annotations from datetime import datetime, timedelta -from typing import Any, Callable, Dict, FrozenSet, Optional +from typing import Any, Callable, Optional from uuid import UUID, uuid4 import attr @@ -19,14 +19,14 @@ class Task: state: Any = None misfire_grace_time: Optional[timedelta] = attr.field(eq=False, order=False, default=None) - def marshal(self, serializer: abc.Serializer) -> Dict[str, Any]: + def marshal(self, serializer: abc.Serializer) -> dict[str, Any]: marshalled = attr.asdict(self) marshalled['func'] = callable_to_ref(self.func) marshalled['state'] = serializer.serialize(self.state) if self.state else None return marshalled @classmethod - def unmarshal(cls, serializer: abc.Serializer, marshalled: Dict[str, Any]) -> Task: + def unmarshal(cls, serializer: abc.Serializer, marshalled: dict[str, Any]) -> Task: marshalled['func'] = callable_from_ref(marshalled['func']) if marshalled['state'] is not None: marshalled['state'] = serializer.deserialize(marshalled['state']) @@ -40,16 +40,16 @@ class Schedule: task_id: str = attr.field(eq=False, order=False) trigger: abc.Trigger = attr.field(eq=False, order=False) args: tuple = attr.field(eq=False, order=False, default=()) - kwargs: Dict[str, Any] = attr.field(eq=False, order=False, factory=dict) + kwargs: dict[str, Any] = attr.field(eq=False, order=False, factory=dict) coalesce: CoalescePolicy = attr.field(eq=False, order=False, default=CoalescePolicy.latest) misfire_grace_time: Optional[timedelta] = attr.field(eq=False, order=False, default=None) - tags: FrozenSet[str] = attr.field(eq=False, order=False, factory=frozenset) + tags: frozenset[str] = attr.field(eq=False, order=False, factory=frozenset) next_fire_time: Optional[datetime] = attr.field(eq=False, order=False, init=False, default=None) last_fire_time: Optional[datetime] = attr.field(eq=False, order=False, init=False, default=None) - def marshal(self, serializer: abc.Serializer) -> Dict[str, Any]: + def marshal(self, serializer: abc.Serializer) -> dict[str, Any]: marshalled = attr.asdict(self) marshalled['trigger_type'] = serializer.serialize(self.args) marshalled['trigger_data'] = serializer.serialize(self.trigger) @@ -71,14 +71,14 @@ class Job: id: UUID = attr.field(factory=uuid4) task_id: str = attr.field(eq=False, order=False) args: tuple = attr.field(eq=False, order=False, default=()) - kwargs: Dict[str, Any] = attr.field(eq=False, order=False, factory=dict) + kwargs: dict[str, Any] = attr.field(eq=False, order=False, factory=dict) schedule_id: Optional[str] = attr.field(eq=False, order=False, default=None) scheduled_fire_time: Optional[datetime] = attr.field(eq=False, order=False, default=None) start_deadline: Optional[datetime] = attr.field(eq=False, order=False, default=None) - tags: FrozenSet[str] = attr.field(eq=False, order=False, factory=frozenset) + tags: frozenset[str] = attr.field(eq=False, order=False, factory=frozenset) started_at: Optional[datetime] = attr.field(eq=False, order=False, init=False, default=None) - def marshal(self, serializer: abc.Serializer) -> Dict[str, Any]: + def marshal(self, serializer: abc.Serializer) -> dict[str, Any]: marshalled = attr.asdict(self) marshalled['args'] = serializer.serialize(self.args) if self.args else None marshalled['kwargs'] = serializer.serialize(self.kwargs) if self.kwargs else None @@ -86,7 +86,7 @@ class Job: return marshalled @classmethod - def unmarshal(cls, serializer: abc.Serializer, marshalled: Dict[str, Any]) -> Task: + def unmarshal(cls, serializer: abc.Serializer, marshalled: dict[str, Any]) -> Task: for key in ('args', 'kwargs'): if marshalled[key] is not None: marshalled[key] = serializer.deserialize(marshalled[key]) |