Effect's DateTime over the standard library: a date-time is an aware
datetime, a named zone a ZoneInfo, a fixed offset a
datetime.timezone. Naive datetimes are rejected with ValueError.
What the standard library already does well stays there
(astimezone, replace, isoformat, timestamp, comparison).
This module adds the current time from Clock, the
CurrentTimeZone service, elapsed-vs-calendar arithmetic (add),
start_of / end_of / nearest, DST disambiguation
(make_zoned) and RFC 9557 zoned strings.
Classes#
Functions#
add#
add(dt, *, years=0, months=0, weeks=0, days=0, hours=0, minutes=0, seconds=0, milliseconds=0, microseconds=0) -> datetimedt moved by the given units, keeping its zone. Months and years
clamp to the last day of the month (Jan 31 + 1 month = Feb 28/29).
Time units are added one at a time, smallest first, each sum cut to a
whole microsecond toward the epoch (Effect's millisecond setTime).
| Parameter | Type | Description |
|---|---|---|
dt | datetime | |
years | int = 0 | |
months | int = 0 | |
weeks | int = 0 | |
days | int = 0 | |
hours | float = 0 | |
minutes | float = 0 | |
seconds | float = 0 | |
milliseconds | float = 0 | |
microseconds | float = 0 |
Returns datetime
def add(
dt: datetime,
*,
years: int = 0,
months: int = 0,
weeks: int = 0,
days: int = 0,
hours: float = 0,
minutes: float = 0,
seconds: float = 0,
milliseconds: float = 0,
microseconds: float = 0,
) -> datetime:
"""``dt`` moved by the given units, keeping its zone. Months and years
clamp to the last day of the month (Jan 31 + 1 month = Feb 28/29).
Time units are added one at a time, smallest first, each sum cut to a
whole microsecond toward the epoch (Effect's millisecond ``setTime``)."""
zone = zone_of(dt, "dt")
epoch_micros = (dt - _EPOCH) // _MICRO
for amount, micros_per_unit in (
(microseconds, 1),
(milliseconds, 1_000),
(seconds, 1_000_000),
(minutes, 60_000_000),
(hours, 3_600_000_000),
):
if amount:
epoch_micros = _add_truncated(epoch_micros, amount, micros_per_unit)
shifted = (_EPOCH + timedelta(microseconds=epoch_micros)).astimezone(zone)
if not (days or weeks or months or years):
return shifted
def move(wall: datetime) -> datetime:
wall += timedelta(days=days + 7 * weeks)
if months:
wall = _with_month(wall, wall.year * 12 + wall.month - 1 + months)
if years:
wall = _with_month(wall, (wall.year + years) * 12 + wall.month - 1)
return wall
return _on_wall_clock(shifted, zone, move)end_of#
end_of(dt, unit, *, week_starts_on=calendar.SUNDAY) -> datetimeThe last microsecond of dt's unit, on its wall clock
(Effect's last millisecond, at datetime's resolution).
| Parameter | Type | Description |
|---|---|---|
dt | datetime | |
unit | Unit | |
week_starts_on | calendar.Day = calendar.SUNDAY |
Returns datetime
def end_of(dt: datetime, unit: Unit, *, week_starts_on: calendar.Day = calendar.SUNDAY) -> datetime:
"""The last microsecond of ``dt``'s ``unit``, on its wall clock
(Effect's last millisecond, at datetime's resolution)."""
return _on_wall_clock(dt, zone_of(dt, "dt"), lambda wall: _end(wall, unit, week_starts_on))nearest#
nearest(dt, unit, *, week_starts_on=calendar.SUNDAY) -> datetimeThe start of dt's unit or of the next one, whichever is
closer on the wall clock; halfway rounds up.
| Parameter | Type | Description |
|---|---|---|
dt | datetime | |
unit | Unit | |
week_starts_on | calendar.Day = calendar.SUNDAY |
Returns datetime
def nearest(
dt: datetime, unit: Unit, *, week_starts_on: calendar.Day = calendar.SUNDAY
) -> datetime:
"""The start of ``dt``'s ``unit`` or of the next one, whichever is
closer on the wall clock; halfway rounds up."""
zone = zone_of(dt, "dt")
def round_wall(wall: datetime) -> datetime:
start = _start(wall, unit, week_starts_on)
after = _end(wall, unit, week_starts_on) + timedelta(microseconds=1)
return start if wall - start < after - wall else after
return _on_wall_clock(dt, zone, round_wall)start_of#
start_of(dt, unit, *, week_starts_on=calendar.SUNDAY) -> datetimeThe first microsecond of dt's unit, on its wall clock.
| Parameter | Type | Description |
|---|---|---|
dt | datetime | |
unit | Unit | |
week_starts_on | calendar.Day = calendar.SUNDAY |
Returns datetime
def start_of(
dt: datetime, unit: Unit, *, week_starts_on: calendar.Day = calendar.SUNDAY
) -> datetime:
"""The first microsecond of ``dt``'s ``unit``, on its wall clock."""
return _on_wall_clock(dt, zone_of(dt, "dt"), lambda wall: _start(wall, unit, week_starts_on))subtract#
subtract(dt, *, years=0, months=0, weeks=0, days=0, hours=0, minutes=0, seconds=0, milliseconds=0, microseconds=0) -> datetimeadd with every unit negated.
| Parameter | Type | Description |
|---|---|---|
dt | datetime | |
years | int = 0 | |
months | int = 0 | |
weeks | int = 0 | |
days | int = 0 | |
hours | float = 0 | |
minutes | float = 0 | |
seconds | float = 0 | |
milliseconds | float = 0 | |
microseconds | float = 0 |
Returns datetime
def subtract(
dt: datetime,
*,
years: int = 0,
months: int = 0,
weeks: int = 0,
days: int = 0,
hours: float = 0,
minutes: float = 0,
seconds: float = 0,
milliseconds: float = 0,
microseconds: float = 0,
) -> datetime:
"""``add`` with every unit negated."""
return add(
dt,
years=-years,
months=-months,
weeks=-weeks,
days=-days,
hours=-hours,
minutes=-minutes,
seconds=-seconds,
milliseconds=-milliseconds,
microseconds=-microseconds,
)is_future#
is_future(dt) -> Effect[bool]Whether dt is after the Clock's current time.
| Parameter | Type | Description |
|---|---|---|
dt | datetime |
Returns Effect[bool]
def is_future(dt: datetime) -> Effect[bool]:
"""Whether ``dt`` is after the ``Clock``'s current time."""
zone_of(dt, "dt")
return now().map(lambda current: current < dt)is_past#
is_past(dt) -> Effect[bool]Whether dt is before the Clock's current time.
| Parameter | Type | Description |
|---|---|---|
dt | datetime |
Returns Effect[bool]
def is_past(dt: datetime) -> Effect[bool]:
"""Whether ``dt`` is before the ``Clock``'s current time."""
zone_of(dt, "dt")
return now().map(lambda current: current > dt)layer_current_zone#
layer_current_zone(zone) -> Layer[CurrentTimeZone]| Parameter | Type | Description |
|---|---|---|
zone | tzinfo |
Returns Layer[CurrentTimeZone]
def layer_current_zone(zone: tzinfo) -> Layer[CurrentTimeZone]:
return layer_succeed(CurrentTimeZone, zone)layer_current_zone_local#
layer_current_zone_local() -> Layer[CurrentTimeZone]Returns Layer[CurrentTimeZone]
def layer_current_zone_local() -> Layer[CurrentTimeZone]:
return layer_sync(CurrentTimeZone, zone_local)layer_current_zone_named#
layer_current_zone_named(zone_id) -> Layer[CurrentTimeZone, IllegalArgumentError]| Parameter | Type | Description |
|---|---|---|
zone_id | str |
Returns Layer[CurrentTimeZone, IllegalArgumentError]
def layer_current_zone_named(zone_id: str) -> Layer[CurrentTimeZone, IllegalArgumentError]:
return layer_effect(CurrentTimeZone, zone_named_effect(zone_id))layer_current_zone_offset#
layer_current_zone_offset(offset) -> Layer[CurrentTimeZone]| Parameter | Type | Description |
|---|---|---|
offset | Duration |
Returns Layer[CurrentTimeZone]
def layer_current_zone_offset(offset: Duration) -> Layer[CurrentTimeZone]:
return layer_succeed(CurrentTimeZone, _offset_zone(offset))now#
now() -> Effect[datetime]The Clock's wall time as an aware UTC datetime (to the
microsecond).
Returns Effect[datetime]
def now() -> Effect[datetime]:
"""The ``Clock``'s wall time as an aware UTC datetime (to the
microsecond)."""
return Clock().map(lambda c: _EPOCH + timedelta(microseconds=c.unsafe_now_nanos() // 1_000))now_in_current_zone#
now_in_current_zone() -> Effect[datetime, Never, CurrentTimeZone]now() in the CurrentTimeZone.
Returns Effect[datetime, Never, CurrentTimeZone]
def now_in_current_zone() -> Effect[datetime, Never, CurrentTimeZone]:
"""``now()`` in the ``CurrentTimeZone``."""
return now().flat_map(set_zone_current)set_zone_current#
set_zone_current(dt) -> Effect[datetime, Never, CurrentTimeZone]The same instant as dt, in the CurrentTimeZone.
| Parameter | Type | Description |
|---|---|---|
dt | datetime |
Returns Effect[datetime, Never, CurrentTimeZone]
def set_zone_current(dt: datetime) -> Effect[datetime, Never, CurrentTimeZone]:
"""The same instant as ``dt``, in the ``CurrentTimeZone``."""
zone_of(dt, "dt")
return CurrentTimeZone().map(dt.astimezone)with_current_zone#
with_current_zone(zone) -> _ProvideService[tzinfo, CurrentTimeZone]Provide zone as the CurrentTimeZone. Use with .pipe.
| Parameter | Type | Description |
|---|---|---|
zone | tzinfo |
Returns _ProvideService[tzinfo, CurrentTimeZone]
def with_current_zone(zone: tzinfo) -> _ProvideService[tzinfo, CurrentTimeZone]:
"""Provide ``zone`` as the ``CurrentTimeZone``. Use with ``.pipe``."""
return provide_service(CurrentTimeZone, zone)with_current_zone_local#
with_current_zone_local() -> _ProvideServiceEffect[CurrentTimeZone, Never, Never]Provide the system's zone (read when the effect runs) as the
CurrentTimeZone. Use with .pipe.
Returns _ProvideServiceEffect[CurrentTimeZone, Never, Never]
def with_current_zone_local() -> _ProvideServiceEffect[CurrentTimeZone, Never, Never]:
"""Provide the system's zone (read when the effect runs) as the
``CurrentTimeZone``. Use with ``.pipe``."""
return provide_service_effect(CurrentTimeZone, sync(zone_local))with_current_zone_named#
with_current_zone_named(zone_id) -> _ProvideServiceEffect[CurrentTimeZone, IllegalArgumentError, Never]Provide the IANA zone zone_id as the CurrentTimeZone; the
effect fails with IllegalArgumentError if there is no such zone.
Use with .pipe.
| Parameter | Type | Description |
|---|---|---|
zone_id | str |
Returns _ProvideServiceEffect[CurrentTimeZone, IllegalArgumentError, Never]
def with_current_zone_named(
zone_id: str,
) -> _ProvideServiceEffect[CurrentTimeZone, IllegalArgumentError, Never]:
"""Provide the IANA zone ``zone_id`` as the ``CurrentTimeZone``; the
effect fails with ``IllegalArgumentError`` if there is no such zone.
Use with ``.pipe``."""
return provide_service_effect(CurrentTimeZone, zone_named_effect(zone_id))with_current_zone_offset#
with_current_zone_offset(offset) -> _ProvideService[tzinfo, CurrentTimeZone]Provide the fixed offset offset from UTC as the
CurrentTimeZone. Use with .pipe.
| Parameter | Type | Description |
|---|---|---|
offset | Duration |
Returns _ProvideService[tzinfo, CurrentTimeZone]
def with_current_zone_offset(offset: Duration) -> _ProvideService[tzinfo, CurrentTimeZone]:
"""Provide the fixed offset ``offset`` from UTC as the
``CurrentTimeZone``. Use with ``.pipe``."""
return with_current_zone(_offset_zone(offset))zone_named_effect#
zone_named_effect(zone_id) -> Effect[tzinfo, IllegalArgumentError]The IANA zone zone_id; fails with IllegalArgumentError if
there is none.
| Parameter | Type | Description |
|---|---|---|
zone_id | str |
Returns Effect[tzinfo, IllegalArgumentError]
def zone_named_effect(zone_id: str) -> Effect[tzinfo, IllegalArgumentError]:
"""The IANA zone ``zone_id``; fails with ``IllegalArgumentError`` if
there is none."""
def check() -> Effect[tzinfo, IllegalArgumentError]:
zone = zone_named(zone_id)
if zone is None:
return fail(IllegalArgumentError(f"Invalid time zone: {zone_id}"))
return succeed(zone)
return suspend(check)format_iso_zoned#
format_iso_zoned(dt) -> strdt.isoformat(), followed by [zone] when dt's zone is a
named ZoneInfo.
| Parameter | Type | Description |
|---|---|---|
dt | datetime |
Returns str
def format_iso_zoned(dt: datetime) -> str:
"""``dt.isoformat()``, followed by ``[zone]`` when ``dt``'s zone is a
named ``ZoneInfo``."""
zone = zone_of(dt, "dt")
key = zone.key if isinstance(zone, ZoneInfo) else None
return dt.isoformat() if key is None else f"{dt.isoformat()}[{key}]"make_zoned_from_string#
make_zoned_from_string(text) -> datetime | NoneParse format_iso_zoned's output, or an ISO string ending in an
offset (whose zone is then that fixed offset). The instant comes from
the ISO part (UTC when it has no offset); None if either part
does not parse.
| Parameter | Type | Description |
|---|---|---|
text | str |
Returns datetime | None
def make_zoned_from_string(text: str) -> datetime | None:
"""Parse ``format_iso_zoned``'s output, or an ISO string ending in an
offset (whose zone is then that fixed offset). The instant comes from
the ISO part (UTC when it has no offset); ``None`` if either part
does not parse."""
found = _ZONED.match(text)
zone: tzinfo | None
if found is None:
iso = text
offset = parse_offset(text)
zone = None if offset is None else zone_offset(offset)
else:
iso, zone = found[1], zone_from_string(found[2])
instant = _parse_instant(iso)
return None if zone is None or instant is None else instant.astimezone(zone)make_zoned#
make_zoned(wall, zone, *, disambiguation='compatible') -> datetime | NoneThe instant whose wall-clock time in zone is wall (a naive
datetime), expressed in zone. None only for a gap or an
overlap under disambiguation="reject".
| Parameter | Type | Description |
|---|---|---|
wall | datetime | |
zone | tzinfo | |
disambiguation | Disambiguation = 'compatible' |
Returns datetime | None
def make_zoned(
wall: datetime, zone: tzinfo, *, disambiguation: Disambiguation = "compatible"
) -> datetime | None:
"""The instant whose wall-clock time in ``zone`` is ``wall`` (a naive
datetime), expressed in ``zone``. ``None`` only for a gap or an
overlap under ``disambiguation="reject"``."""
if wall.tzinfo is not None:
raise ValueError(f"wall must be a naive datetime (a wall-clock time), got {wall!r}")
first = wall.replace(tzinfo=zone, fold=0).astimezone(UTC)
second = wall.replace(tzinfo=zone, fold=1).astimezone(UTC)
if first == second:
return first.astimezone(zone)
if disambiguation == "reject":
return None
# PEP 495: in an overlap fold=0 is the earlier instant; in a gap fold=0
# reads the wall time with the offset from before the gap, which gives
# the later instant.
earlier, later = sorted((first, second))
if disambiguation == "earlier":
return earlier.astimezone(zone)
if disambiguation == "later":
return later.astimezone(zone)
in_gap = first.astimezone(zone).replace(tzinfo=None) != wall
return (later if in_gap else earlier).astimezone(zone)zone_from_string#
zone_from_string(text) -> tzinfo | NoneA zone from an offset (+05:30, -0530, GMT+05:30) or an
IANA name (Europe/London, GMT).
| Parameter | Type | Description |
|---|---|---|
text | str |
Returns tzinfo | None
def zone_from_string(text: str) -> tzinfo | None:
"""A zone from an offset (``+05:30``, ``-0530``, ``GMT+05:30``) or an
IANA name (``Europe/London``, ``GMT``)."""
if _OFFSET_ZONE.match(text):
offset = parse_offset(text)
return None if offset is None else zone_offset(offset)
return zone_named(text)zone_local#
zone_local() -> tzinfoThe system's zone: $TZ, else /etc/localtime (named when it
links into a zoneinfo tree), else the current fixed offset.
Returns tzinfo
def zone_local() -> tzinfo:
"""The system's zone: ``$TZ``, else ``/etc/localtime`` (named when it
links into a zoneinfo tree), else the current fixed offset."""
for key in (os.environ.get("TZ", "").removeprefix(":"), _localtime_key()):
zone = zone_named(key) if key else None
if zone is not None:
return zone
try:
with open(_LOCALTIME, "rb") as file:
return ZoneInfo.from_file(file)
except (OSError, ValueError):
return datetime.now().astimezone().tzinfo or UTCzone_named#
zone_named(zone_id) -> ZoneInfo | NoneThe IANA zone zone_id, or None if there is no such zone.
| Parameter | Type | Description |
|---|---|---|
zone_id | str |
Returns ZoneInfo | None
def zone_named(zone_id: str) -> ZoneInfo | None:
"""The IANA zone ``zone_id``, or ``None`` if there is no such zone."""
try:
return ZoneInfo(zone_id)
except (ZoneInfoNotFoundError, ValueError):
return Nonezone_offset#
zone_offset(offset) -> timezone | NoneA fixed offset from UTC, or None outside timezone's range
(strictly within a day).
| Parameter | Type | Description |
|---|---|---|
offset | timedelta |
Returns timezone | None
def zone_offset(offset: timedelta) -> timezone | None:
"""A fixed offset from UTC, or ``None`` outside ``timezone``'s range
(strictly within a day)."""
try:
return timezone(offset)
except ValueError:
return NoneOn This Page
ClassesFunctionsaddend_ofneareststart_ofsubtractis_futureis_pastlayer_current_zonelayer_current_zone_locallayer_current_zone_namedlayer_current_zone_offsetnownow_in_current_zoneset_zone_currentwith_current_zonewith_current_zone_localwith_current_zone_namedwith_current_zone_offsetzone_named_effectformat_iso_zonedmake_zoned_from_stringmake_zonedzone_from_stringzone_localzone_namedzone_offset