Skip to content

SSF Tools - Timestamp Service Architecture

Overview

The Timestamp Service provides a unified, protocol-based approach to datetime operations across all SSF Tools commands. This architecture enables consistent temporal data handling while maintaining loose coupling through dependency injection patterns and supporting various timestamp operations including formatting, parsing, timezone conversion, and performance measurement.

Architectural Principles

1. Protocol-Based Design

All timestamp services implement well-defined protocols, enabling: - Dependency injection: Services receive timestamp interfaces, not concrete implementations - Testability: Easy mocking and stubbing for unit tests with fixed times - Flexibility: Swappable implementations (UTC, local time, mock time) - Type safety: MyPy-compliant interfaces with clear contracts - Composability: Small, focused protocols that can be combined

2. Separation of Concerns

Each timestamp operation is handled through a dedicated interface: - Time Generation: Current time, UTC time, timezone-aware timestamps - Formatting: ISO 8601, RFC 3339, custom format strings - Parsing: String to datetime conversion with timezone handling - Conversion: Timezone conversion, Unix timestamp conversion - Measurement: Duration calculation, elapsed time tracking

3. Consistent Temporal Handling

All timestamp services implement consistent temporal patterns: - UTC by default: All internal operations use UTC timestamps - Timezone awareness: Proper timezone handling and conversion - Immutable operations: No side effects on datetime objects - Type safety: Clear return types for all operations

Architecture Overview

The Timestamp Service follows a layered architecture with clear separation of concerns:

Service Composition

graph TD Client[Client Code] --> TS[TimestampService] TS --> DT[DateTime Operations] TS --> TZ[Timezone Manager] TS --> FT[Formatters] TS --> PR[Parsers] TS --> MC[Measurement Calculations] TS --> DC[Duration Calculator] TS --> EC[Elapsed Time Counter] classDef service fill:#e1f5fe,stroke:#0277bd,stroke-width:2px classDef component fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px classDef feature fill:#e8f5e8,stroke:#2e7d32,stroke-width:2px class TS service class DT,TZ,FT,PR component class MC,DC,EC feature

Core Protocols

graph LR subgraph "Time Operations" TP["TimestampProtocol<br/>- now()<br/>- utc_now()<br/>- from_timestamp()"] TO["TimeOperations<br/>- add_seconds()<br/>- elapsed_seconds()"] end subgraph "Format Operations" FP["FormatProtocol<br/>- format_iso()<br/>- format_rfc3339()"] PP["ParseProtocol<br/>- parse_iso()<br/>- parse_rfc3339()"] end subgraph "Conversion Operations" CP["ConversionProtocol<br/>- to_utc()<br/>- to_timestamp()"] ZP["ZoneProtocol<br/>- timezone handling"] end classDef protocol fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px class TP,TO,FP,PP,CP,ZP protocol

Implementation Architecture

graph TD subgraph "Timestamp Implementation" TS[TimestampService] -.-> TSI[TimestampService Implementation] end subgraph "Time Management" TM[TimeManager] -.-> TMI[DateTime Implementation] TZ[TimezoneManager] -.-> TZI[Timezone Implementation] end subgraph "Format Handlers" FH[FormatHandler] -.-> FHI[ISO/RFC Format Implementation] PH[ParseHandler] -.-> PHI[String Parser Implementation] end classDef protocol fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px classDef implementation fill:#e8f5e8,stroke:#2e7d32,stroke-width:2px class TS,TM,TZ,FH,PH protocol class TSI,TMI,TZI,FHI,PHI implementation

Dependency Injection

graph TB CC[Core Container] --> TS[TimestampService] TS --> PUS[PathUtilitiesService] TS --> CS[ConfigurationService] TS --> MS[MetricsService] TS --> LS[LoggingService] CC --> TZ[Timezone Config] TZ --> TS classDef container fill:#fff3e0,stroke:#ef6c00,stroke-width:2px classDef service fill:#e1f5fe,stroke:#0277bd,stroke-width:2px classDef config fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px classDef dependent fill:#e8f5e8,stroke:#2e7d32,stroke-width:2px class CC container class TS service class TZ config class PUS,CS,MS,LS dependent

Core Components

The Timestamp Service implements modular components that work together to provide comprehensive temporal functionality:

Timestamp Service Implementation

kp_ssf_tools.core.services.timestamp.service.TimestampService

Timestamp service implementation providing datetime operations.

Functions
__init__(default_timezone=UTC)

Initialize timestamp service.

Parameters:

Name Type Description Default
default_timezone timezone

Default timezone for operations (defaults to UTC)

 UTC
add_seconds(dt, seconds)

Add seconds to a datetime.

Parameters:

Name Type Description Default
dt datetime

Base datetime

 required
seconds float

Seconds to add (can be negative)

 required

Returns:

Type Description
datetime

Modified datetime
elapsed_seconds(start_time, end_time=None)

Calculate elapsed seconds between two datetimes.

Parameters:

Name Type Description Default
start_time datetime

Start datetime

 required
end_time datetime | None

End datetime (defaults to current UTC time)

 None

Returns:

Type Description
float

Elapsed seconds as float
format_filename(dt)

Format datetime for use in filenames (YYYYMMDD-HHMMSS).

Parameters:

Name Type Description Default
dt datetime

Datetime to format

 required

Returns:

Type Description
str

Filename-safe formatted string (YYYYMMDD-HHMMSS)
format_filename_now()

Format current local time for use in filenames (YYYYMMDD-HHMMSS).

The timestamp is cached on first call and reused for subsequent calls during the same application run to ensure consistent timestamps across all files in a batch processing operation.

Returns:

Type Description
str

Filename-safe formatted string for current local time (cached)
format_iso(dt)

Format datetime as ISO 8601 string.

Parameters:

Name Type Description Default
dt datetime

Datetime to format

 required

Returns:

Type Description
str

ISO 8601 formatted string
format_rfc3339(dt)

Format datetime as RFC 3339 string.

Parameters:

Name Type Description Default
dt datetime

Datetime to format

 required

Returns:

Type Description
str

RFC 3339 formatted string
from_timestamp(timestamp)

Create datetime from Unix timestamp.

Parameters:

Name Type Description Default
timestamp float

Unix timestamp (seconds since epoch)

 required

Returns:

Type Description
datetime

Datetime object in UTC
now()

Get current local datetime.

parse_iso(iso_string)

Parse ISO 8601 string to datetime.

Parameters:

Name Type Description Default
iso_string str

ISO 8601 formatted string

 required

Returns:

Type Description
datetime

Parsed datetime object

Raises:

Type Description
ValueError

If string cannot be parsed
reset_filename_cache()

Reset the cached filename timestamp to force regeneration on next call.

This is useful for testing or if you want to force a new timestamp during the same application run.

to_timestamp(dt)

Convert datetime to Unix timestamp.

Parameters:

Name Type Description Default
dt datetime

Datetime to convert

 required

Returns:

Type Description
float

Unix timestamp (seconds since epoch)
to_utc(dt)

Convert datetime to UTC.

Parameters:

Name Type Description Default
dt datetime

Datetime to convert

 required

Returns:

Type Description
datetime

UTC datetime
utc_now()

Get current UTC datetime.

Core Protocols

kp_ssf_tools.core.services.timestamp.interfaces.TimestampProtocol

Bases: Protocol

Protocol defining the timestamp service interface for dependency injection.

Functions
add_seconds(dt, seconds)

Add seconds to a datetime.

elapsed_seconds(start_time, end_time=None)

Calculate elapsed seconds between two datetimes.

format_filename(dt)

Format datetime for use in filenames (YYYYMMDD-HHMMSS).

format_filename_now()

Format current local time for use in filenames (YYYYMMDD-HHMMSS).

format_iso(dt)

Format datetime as ISO 8601 string.

format_rfc3339(dt)

Format datetime as RFC 3339 string.

from_timestamp(timestamp)

Create datetime from Unix timestamp.

now()

Get current local datetime.

parse_iso(iso_string)

Parse ISO 8601 string to datetime.

reset_filename_cache()

Reset the cached filename timestamp to force regeneration on next call.

to_timestamp(dt)

Convert datetime to Unix timestamp.

to_utc(dt)

Convert datetime to UTC.

utc_now()

Get current UTC datetime.

Timestamp Service Workflow

Time Generation Sequence

sequenceDiagram participant Client participant TS as TimestampService participant TZ as TimezoneManager participant DT as DateTime Client->>TS: request current time TS->>TZ: get timezone config TZ-->>TS: timezone settings TS->>DT: generate timestamp DT-->>TS: raw datetime TS->>TS: apply timezone TS-->>Client: formatted timestamp

Format and Parse Operations

sequenceDiagram participant Client participant TS as TimestampService participant FH as FormatHandler participant PH as ParseHandler Note over Client,PH: Format Operation Client->>TS: format_iso(datetime) TS->>FH: apply ISO format FH-->>TS: formatted string TS-->>Client: ISO 8601 string Note over Client,PH: Parse Operation Client->>TS: parse_rfc3339(string) TS->>PH: parse RFC format PH-->>TS: datetime object TS-->>Client: timezone-aware datetime

Duration and Measurement

sequenceDiagram participant Client participant TS as TimestampService participant MC as MeasurementCalculator participant DC as DurationCalculator Client->>TS: start_measurement() TS->>MC: create start point MC-->>TS: start timestamp Note right of Client: Process execution... Client->>TS: end_measurement() TS->>DC: calculate duration DC->>MC: get start timestamp MC-->>DC: start time DC-->>TS: elapsed duration TS-->>Client: duration metrics

Usage Patterns

Basic Dependency Injection

Commands and services receive the timestamp service through constructor injection:

from kp_ssf_tools.core.services.timestamp.interfaces import TimestampProtocol

class DataProcessor:
    def __init__(self, timestamp_service: TimestampProtocol):
        self._timestamp_service = timestamp_service

    def process_data(self, data: bytes) -> ProcessResult:
        start_time = self._timestamp_service.utc_now()

        # Process data...

        end_time = self._timestamp_service.utc_now()
        duration = end_time - start_time

        return ProcessResult(
            processed_at=self._timestamp_service.format_iso(end_time),
            duration_seconds=duration.total_seconds()
        )

Performance Measurement

Timestamp service provides standardized performance measurement capabilities:

class CommandExecutor:
    def __init__(self, timestamp_service: TimestampProtocol):
        self._timestamp = timestamp_service

    def execute_with_timing(self, command: str) -> ExecutionResult:
        start = self._timestamp.utc_now()

        try:
            result = self._execute_command(command)
            success = True
        except Exception as e:
            result = str(e)
            success = False
        finally:
            end = self._timestamp.utc_now()
            duration = end - start

        return ExecutionResult(
            command=command,
            result=result,
            success=success,
            started_at=self._timestamp.format_rfc3339(start),
            completed_at=self._timestamp.format_rfc3339(end),
            duration_ms=duration.total_seconds() * 1000
        )

Timezone Handling

The service provides consistent timezone management across different environments:

class LogProcessor:
    def __init__(self, timestamp_service: TimestampProtocol):
        self._timestamp = timestamp_service

    def normalize_log_timestamps(self, log_entries: list[dict]) -> list[dict]:
        """Convert all log timestamps to UTC for consistent processing."""
        normalized = []

        for entry in log_entries:
            # Parse timestamp from various formats
            if 'iso_timestamp' in entry:
                dt = self._timestamp.parse_iso(entry['iso_timestamp'])
            elif 'unix_timestamp' in entry:
                dt = self._timestamp.from_timestamp(entry['unix_timestamp'])
            else:
                dt = self._timestamp.utc_now()  # Fallback

            # Convert to UTC and format consistently
            utc_dt = self._timestamp.to_utc(dt)
            entry['normalized_timestamp'] = self._timestamp.format_iso(utc_dt)
            normalized.append(entry)

        return normalized

Container Configuration

The Timestamp Service integrates with the core container system for dependency injection:

Service Registration

# In kp_ssf_tools.containers.core_container
from dependency_injector import containers, providers
from kp_ssf_tools.core.services.timestamp import TimestampService

class CoreContainer(containers.DeclarativeContainer):
    # Configuration
    timestamp_config = providers.Configuration()

    # Timestamp Service
    timestamp_service = providers.Singleton(
        TimestampService,
        config=timestamp_config,
        default_timezone='UTC',
        iso_format_precision='milliseconds'
    )

    # Example: Data Processor with timestamp dependency
    data_processor = providers.Factory(
        DataProcessor,
        timestamp_service=timestamp_service
    )

Configuration Structure

# ssf-tools-config.yaml
timestamp:
  default_timezone: "UTC"
  format_precision: "milliseconds"
  enable_performance_tracking: true
  measurement_precision: "microseconds"
  locale_settings:
    date_format: "ISO"
    time_format: "24hour"

Service Dependencies

The Timestamp Service can integrate with other core services:

class EnhancedTimestampService(TimestampService):
    def __init__(
        self,
        config_service: ConfigurationProtocol,
        logging_service: LoggingProtocol,
        metrics_service: MetricsProtocol
    ):
        super().__init__()
        self._config = config_service
        self._logger = logging_service
        self._metrics = metrics_service

    def performance_measurement(self, operation_name: str):
        """Context manager for performance measurement with metrics integration."""
        return PerformanceContext(
            operation_name=operation_name,
            timestamp_service=self,
            metrics_service=self._metrics,
            logger=self._logger
        )

Performance Considerations

Timezone Caching

The service implements timezone caching to avoid repeated timezone lookups:

  • Timezone objects: Cached for frequently used timezones
  • Format patterns: Compiled regex patterns cached for parsing
  • Configuration: Timezone settings cached per container lifecycle

Memory Efficiency

  • Immutable operations: No datetime object mutation
  • Lazy evaluation: Timezone conversions only when needed
  • Pooled formatters: Reusable formatter objects

Precision Settings

Different precision levels available based on use case:

  • Second precision: Standard operations, file timestamps
  • Millisecond precision: Performance measurements, logging
  • Microsecond precision: High-precision timing, benchmarking
  • Nanosecond precision: Ultra-precise measurements (where supported)

The Timestamp Service provides a robust foundation for all temporal operations in SSF Tools, ensuring consistency, testability, and performance across the entire system.