`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, *, _shutdown_mode: bool = False, )¶

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.
_shutdown_mode – Internal parameter. If True, creates a minimal no-op instance that skips config reads. This is used during interpreter shutdown to avoid side effects.

classmethod clear_global_shutdown() → None¶: Clear global shutdown mode.

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 global_shutdown_context() → Iterator[None]¶

Context manager for global shutdown mode.

Disables writes on existing instance while still queuing notifications. On exit, stops and clears the instance, then clears shutdown mode.

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.

classmethod set_global_shutdown() → None¶

Enter global shutdown mode.

After this call:

Existing instance (if any) has writes disabled (but notifications still queue writes)
New instance() calls return a minimal no-op instance that skips config reads

Writes that are queued during shutdown will be flushed when stop() is called.

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`¶