tlc.core.objects.tables.system_tables.timestamp_helper

Module Contents

Classes

Class	Description
PendingWrite	Represents a pending timestamp write operation.
IndexTimestampModel
TimestampHelper	Helper class for managing timestamp operations with debouncing support.

API

class tlc.core.objects.tables.system_tables.timestamp_helper.PendingWrite

Represents a pending timestamp write operation.

Parameters:

• scheduled_time – Monotonic time when write should occur

• current_interval – Current debounce interval for this URL

• write_count – Number of writes scheduled for this URL

scheduled_time: float = None

write_count: int = 1

backoff_level: int = 0

class tlc.core.objects.tables.system_tables.timestamp_helper.IndexTimestampModel(/, **data: typing.Any)

Bases: pydantic.BaseModel

timestamp: datetime.datetime = None

serialize_timestamp(dt: datetime.datetime) → str

classmethod ensure_timezone(data: dict) → dict

Ensures timestamp has timezone information, defaulting to UTC if none provided. Also handles parsing string timestamps.

Parameters:

data – Input data dictionary

Returns:

Validated data dictionary

class tlc.core.objects.tables.system_tables.timestamp_helper.TimestampHelper(debounce_interval: float | None = None, debounce_backoff_threshold: int | None = None, debounce_backoff_multiplier: float | None = None, debounce_backoff_max_level: int | None = None)

Helper class for managing timestamp operations with debouncing support.

Provides debounced writes to prevent cascading updates to the same URL within a specified time window. The debounce interval dynamically increases with multiple writes up to a maximum value.

Note: It is safe to shutdown and restart the TimestampHelper the writer thread will be restarted automatically.

Initialize the TimestampHelper.

Parameters:

• debounce_interval – Initial debounce interval in seconds

• debounce_backoff_threshold – Threshold for increasing backoff level

• debounce_backoff_multiplier – Multiplier for increasing the interval

• debounce_backoff_max_level – Maximum backoff level allowed

timestamp_helper_instance: tlc.core.objects.tables.system_tables.timestamp_helper.TimestampHelper | None = None

property debounce_interval: float

Get the current debounce interval in seconds

property debounce_backoff_multiplier: float

Get the current backoff multiplier

property debounce_backoff_max_level: int

Get the maximum backoff level

property debounce_backoff_threshold: int

Get the current backoff threshold

classmethod instance() → tlc.core.objects.tables.system_tables.timestamp_helper.TimestampHelper

property is_running: bool

Check if the background debounce write thread is running.

schedule_timestamp_write(url: tlc.core.url.Url) → None

Schedule a timestamp write for the given URL with dynamic debouncing. The debounce interval increases with multiple writes up to the maximum value.

Parameters:

url – URL to write the timestamp for

start() → None

Start the Timestamp debounce writer thread

The write thread starts automatically the first time. After this it must be explicitly started whenever it is stopped.

Whenever the timestamp write thread is explicitly stopped it needs to be explicitly restarted.

flush() → None

Flush all pending writes immediately. This ensures all pending timestamps are written immediately.

stop() → None

Flush all pending writes and stop the Timestamp debounce writer thread

read_timestamp(url: tlc.core.url.Url) → datetime.datetime

Read a timestamp from the given URL and return it as a datetime object

This expects the timestamp content to be in ISO 8601 format with a timezone offset, if no timezone offset is provided it will be converted to UTC.

remove_timestamp(url: tlc.core.url.Url) → None

Remove the timestamp at the given URL

This can be used to signal that content has changed and indexing should be re-run. Only updates internal state if the delete operation succeeds.

Parameters:

url – URL to remove the timestamp for

Raises:

OSError – If deletion fails

get_written_timestamps() → dict[tlc.core.url.Url, datetime.datetime]

Get all currently written timestamps as a dict with URL as key and timestamp as value.

Returns:

Dictionary mapping URLs to their timestamps

temporary_debounce_interval(interval: float) → collections.abc.Iterator[None]

Context manager that temporarily sets a different debounce interval.

The original interval is restored when exiting the context, even if an exception occurs. Thread-safe using the existing read-write lock.

Parameters:

interval – The temporary debounce interval in seconds to use

Raises:

ValueError – If interval is not positive

disabled_timestamp_writes() → collections.abc.Iterator[None]

Context manager that temporarily disables all timestamp writes.

Any pending writes are flushed before disabling writes. New writes will be queued but not executed until after exiting the context. Thread-safe using the existing read-write lock.

Example usage: with helper.disabled_timestamp_writes(): # No timestamps will be written during this block …