`tlc.core.objects.tables.system_tables.timestamp_helper`¶

Module Contents¶

Classes¶

Class	Description
`IndexTimestampModel`	Pydantic model for a timestamp index entry.
`PendingWrite`	Represents a pending timestamp write operation.
`TimestampHelper`	Helper class for managing timestamp operations with debouncing support.
`UrlWriteState`	Tracks the write state for a specific URL.

API¶

class IndexTimestampModel(

/,

**data: typing.Any,

)¶

Bases: pydantic.BaseModel

Pydantic model for a timestamp index entry.

Parameters:: timestamp – The timestamp value.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

classmethod ensure_timezone( data: dict, ) → dict¶

Ensure the timestamp has timezone information, defaulting to UTC if none provided.

Also handles parsing string timestamps.

Parameters:: data – Input data dictionary.
Returns:: Validated data dictionary.

serialize_timestamp( dt: datetime, ) → str¶

Serialize a datetime object to an ISO8601 string in UTC.

Parameters:: dt – The datetime object to serialize.
Returns:: The ISO8601 string representation in UTC.

timestamp: datetime = None¶

class PendingWrite¶

Represents a pending timestamp write operation.

Parameters:

scheduled_time – Monotonic time when write should occur.
url – URL to write the timestamp for.

scheduled_time: float = None¶

url: Url = None¶

class 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.

property debounce_backoff_max_level: int¶

Get the maximum backoff level.

Returns:: The maximum backoff level.

property debounce_backoff_multiplier: float¶

Get the current backoff multiplier.

Returns:: The current backoff multiplier.

property debounce_backoff_threshold: int¶

Get the current backoff threshold.

Returns:: The current backoff threshold.

property debounce_interval: float¶

Get the current debounce interval in seconds.

Returns:: The current debounce interval in seconds.

disabled_timestamp_notifications() → Iterator[None]¶

Context manager that temporarily disables all timestamp notifications.

Pending writes are processed normally but no new notifications are collected and no writes are scheduled.

disabled_timestamp_writes() → Iterator[None]¶

Context manager that temporarily disables all timestamp writes.

Any pending writes are flushed before disabling writes (as a part of the stop operation). 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
    ...

flush() → None¶: Flush all pending writes immediately. This ensures all pending timestamps are written immediately.

get_written_timestamps() → dict[Url, datetime]¶

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

Returns:: Dictionary mapping URLs to their timestamps.

classmethod instance() → TimestampHelper¶

Get the singleton instance of TimestampHelper.

Returns:: The singleton TimestampHelper instance.

property is_running: bool¶

Check if the background debounce write thread is running.

Returns:: True if the write thread is running, False otherwise.

notify_operation( url: Url, op: Literal[write, delete], ) → None¶

Handle notification of a file operation.

Parameters:

url – The URL where the operation occurred.
op – The type of operation performed (“write” or “delete”).

process_timestamp( project_root: Url, ) → None¶

Process a timestamp operation for the given project root.

Parameters:: project_root – The root URL of the project.

read_timestamp( url: Url, ) → 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.

Parameters:: url – The URL to read the timestamp from.
Returns:: The timestamp as a datetime object.

remove_timestamp( 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.

schedule_timestamp_write( url: Url, ) → None¶

Schedule a timestamp write for the given URL.

Multiple calls for the same URL are coalesced - only one write will be queued at a time, but load accumulates to influence backoff timing.

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.

Note. The writer thread is automatically started if it is stopped unless the disabled_timestamp_writes attribute is set.

stop() → None¶: Flush all pending writes and stop the Timestamp debounce writer thread.

temporary_debounce_interval( interval: float, ) → 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.

timestamp_helper_instance: TimestampHelper | None = None¶

class UrlWriteState¶

Tracks the write state for a specific URL.

Parameters:

load – Current load level (decays over time).
last_updated – Last time the load was updated.
queued – Whether this URL is currently in the write queue.

last_updated: float = None¶

load: float = None¶

queued: bool = False¶

tlc.core.objects.tables.system_tables.timestamp_helper¶

Module Contents¶

Classes¶

API¶

`tlc.core.objects.tables.system_tables.timestamp_helper`¶