dodil/erp-brick-fixed-assets

DODIL ERP brick: fixed assets

src/brick_fixed_assets/infrastructure/business_time.py

"""'Today' is the date in business_timezone — ONE value per bucket (shared.yaml conventions.business_timezone).

This brick keeps NO copy of the zone: it reads the one shared value through `SOURCE`, a small port so the read can move
to the platform without touching a use case. Until the platform holds that value (a person with an owner's `<ns>:settings:controls` in any brick sets it, erp-platform
#5) the zone is the BUSINESS_TIMEZONE environment variable, then Europe/London. Tests install their own source.

Which zone is in force is decided in BUSINESS time, not UTC: walking the dated rows in order, a row takes over once the
business date counted in the zone BEFORE it has reached the row's valid_from — so a zone dated tomorrow is never in
force at once (at 18:00 in Pago_Pago the UTC date is already tomorrow).

'Today' never goes backwards: the last business date seen is kept (fa_clock) and a smaller one is refused in favour of it.
"""
import os
from datetime import date
from typing import Callable, Optional
from zoneinfo import ZoneInfo

from erp_platform import work

EPOCH = date(1900, 1, 1)
DEFAULT_ZONE = "Europe/London"


def _bucket_value(conn) -> list:
    """[(valid_from, zone)], oldest first. The bucket-level value is the PLATFORM's (erp-platform #5) and is not there
    yet: until it is, the zone is BUSINESS_TIMEZONE (the deployment's one setting). A failed read must never poison the
    caller's transaction, so nothing is queried from here."""
    return []


# The port: a callable (conn) -> [(valid_from, zone)]. Replace it (tests, or the platform later); never add a table here.
SOURCE: Callable = _bucket_value


def timezone_name(conn) -> str:
    base = os.environ.get("BUSINESS_TIMEZONE") or DEFAULT_ZONE
    name = base
    now = work.now()
    for valid_from, zone in SOURCE(conn):
        if valid_from <= now.astimezone(ZoneInfo(name)).date():
            name = zone or name
        else:
            break
    return name


def today(conn, floor: Optional[date] = None) -> date:
    """The business date now, never below `floor` (the last date this brick already stamped on a row)."""
    day = work.now().astimezone(ZoneInfo(timezone_name(conn))).date()
    return max(day, floor) if floor else day