Cache - tif1

Overview

The tif1 cache system provides:

SQLite backend: Persistent storage with WAL mode for concurrency
In-memory LRU cache: Fast access for frequently used data
Thread-safe operations: Lock-free reads, protected writes
Async support: Asynchronous cache operations
Specialized telemetry cache: Optimized for telemetry data

Functions

get_cache()

def get_cache() -> Cache

Get the global cache instance (lazy initialization, thread-safe singleton). Returns:

Cache: Global cache instance

Example:

import tif1

cache = tif1.get_cache()
print(cache.cache_dir)

Cache Class

Constructor

class Cache:
    """Cache with SQLite backend and async support."""
    
    def __init__(self, cache_dir: Path | None = None)

Initialize cache with optional custom directory. Parameters:

cache_dir (Path, optional): Cache directory path. If None, uses:
1. TIF1_CACHE_DIR environment variable
2. Configuration value from get_config()
3. OS-specific default (see Default Cache Directory section)

Example:

from pathlib import Path
from tif1.cache import Cache

# Use default location
cache = Cache()

# Use custom location
cache = Cache(cache_dir=Path('~/my-project/.cache'))

Attributes

Attribute	Type	Description
`cache_dir`	Path	Directory where cache is stored
`db_path`	Path	Path to SQLite database file
`read_only`	bool	Whether cache is in read-only mode

Methods

General Cache Operations

get()

def get(key: str) -> Any | None

Get cached data (thread-safe). Uses lock-free memory cache reads for performance. Parameters:

key (str): Cache key to lookup

Returns:

Any | None: Cached data or None if not found

Example:

cache = tif1.get_cache()
data = cache.get('2024/Monaco/R/laps')
if data:
    print('Cache hit!')

get_async()

async def get_async(key: str) -> Any | None

Get cached data asynchronously. Parameters:

key (str): Cache key to lookup

Returns:

Any | None: Cached data or None if not found

Example:

import asyncio

async def fetch_data():
    cache = tif1.get_cache()
    data = await cache.get_async('2024/Monaco/R/laps')
    return data

data = asyncio.run(fetch_data())

set()

def set(key: str, data: Any) -> None

Set cached data (thread-safe). Optimized to update memory cache first (<1ms) before SQLite. Parameters:

key (str): Cache key
data (Any): Data to cache (must be JSON-serializable)

Example:

cache = tif1.get_cache()
cache.set('2024/Monaco/R/laps', laps_data)

set_async()

async def set_async(key: str, data: Any) -> None

Set cached data asynchronously. Parameters:

key (str): Cache key
data (Any): Data to cache

Example:

async def store_data():
    cache = tif1.get_cache()
    await cache.set_async('2024/Monaco/R/laps', laps_data)

Telemetry-Specific Operations

get_telemetry()

def get_telemetry(
    year: int,
    gp: str,
    session: str,
    driver: str,
    lap: int
) -> Any | None

Get cached telemetry data (thread-safe). Uses lock-free memory reads. Parameters:

year (int): Season year
gp (str): Grand Prix identifier
session (str): Session type (e.g., ‘R’, ‘Q’, ‘FP1’)
driver (str): Driver code (e.g., ‘VER’, ‘HAM’)
lap (int): Lap number

Returns:

Any | None: Cached telemetry data or None if not found

Example:

cache = tif1.get_cache()
telemetry = cache.get_telemetry(2024, 'Monaco', 'R', 'VER', 1)
if telemetry:
    print(f"Cached telemetry for VER lap 1")

get_telemetry_async()

async def get_telemetry_async(
    year: int,
    gp: str,
    session: str,
    driver: str,
    lap: int
) -> Any | None

Get cached telemetry data asynchronously. Parameters: Same as get_telemetry() Returns:

Any | None: Cached telemetry data or None if not found

get_telemetry_batch()

def get_telemetry_batch(
    year: int,
    gp: str,
    session: str,
    driver_laps: list[tuple[str, int]]
) -> dict[tuple[str, int], Any]

Get multiple cached telemetry entries in a single batch (thread-safe). Optimized for bulk lookups. Parameters:

year (int): Season year
gp (str): Grand Prix identifier
session (str): Session type
driver_laps (list[tuple[str, int]]): List of (driver_code, lap_number) tuples

Returns:

dict[tuple[str, int], Any]: Dictionary mapping (driver, lap) to telemetry data

Example:

cache = tif1.get_cache()

# Get telemetry for multiple laps at once
driver_laps = [('VER', 1), ('VER', 2), ('HAM', 1)]
results = cache.get_telemetry_batch(2024, 'Monaco', 'R', driver_laps)

for (driver, lap), telemetry in results.items():
    print(f"Got telemetry for {driver} lap {lap}")

get_telemetry_batch_async()

async def get_telemetry_batch_async(
    year: int,
    gp: str,
    session: str,
    driver_laps: list[tuple[str, int]]
) -> dict[tuple[str, int], Any]

Get multiple cached telemetry entries asynchronously in a single batch. Parameters: Same as get_telemetry_batch() Returns:

dict[tuple[str, int], Any]: Dictionary mapping (driver, lap) to telemetry data

set_telemetry()

def set_telemetry(
    year: int,
    gp: str,
    session: str,
    driver: str,
    lap: int,
    data: Any
) -> None

Set cached telemetry data (thread-safe). Optimized to update memory cache first (<1ms). Parameters:

year (int): Season year
gp (str): Grand Prix identifier
session (str): Session type
driver (str): Driver code
lap (int): Lap number
data (Any): Telemetry data to cache

Example:

cache = tif1.get_cache()
cache.set_telemetry(2024, 'Monaco', 'R', 'VER', 1, telemetry_data)

set_telemetry_async()

async def set_telemetry_async(
    year: int,
    gp: str,
    session: str,
    driver: str,
    lap: int,
    data: Any
) -> None

Set cached telemetry data asynchronously. Parameters: Same as set_telemetry()

Cache Management

has_session_data()

def has_session_data(year: int, gp: str, session: str) -> bool

Check if cache contains data for a session (JSON or telemetry). Parameters:

year (int): Season year
gp (str): Grand Prix identifier
session (str): Session type

Returns:

bool: True if session data exists in cache

Example:

cache = tif1.get_cache()
if cache.has_session_data(2024, 'Monaco', 'R'):
    print('Session data is cached')

clear()

def clear() -> None

Clear all cached data (SQLite and memory). Example:

cache = tif1.get_cache()
cache.clear()
print('Cache cleared')

close()

def close() -> None

Close database connection and clear memory caches. Called automatically at exit. Example:

cache = tif1.get_cache()
# ... use cache ...
cache.close()  # Usually not needed, cleanup is automatic

Default Cache Directory

The default cache directory is platform-specific:

Platform	Default Location
Windows	`%LOCALAPPDATA%/Temp/tif1`
macOS	`~/Library/Caches/tif1`
Linux	`~/.cache/tif1` (or `~/.tif1` if `~/.cache` doesn’t exist)

Override with:

TIF1_CACHE_DIR environment variable
Config file setting: cache_dir
Constructor parameter: Cache(cache_dir=...)

Cache Architecture

Two-Tier Caching

In-Memory LRU Cache:
- Lock-free reads for maximum concurrency
- Bounded by memory_cache_max_items (default: 1024)
- Separate cache for telemetry: memory_telemetry_cache_max_items (default: 2048)
- <1ms access time
SQLite Backend:
- Persistent storage with WAL mode for concurrency
- Two tables: cache (general) and telemetry_cache (specialized)
- Batched commits every cache_commit_interval writes (default: 25)
- Thread-safe with fine-grained locking

Thread Safety

Lock-free reads: Memory cache reads use no locks (GIL-protected in CPython)
Separate locks: Different locks for memory cache and SQLite operations
Minimal lock duration: Memory updates (<1ms) separate from SQLite writes

Performance Optimizations

WAL (Write-Ahead Logging) mode for better SQLite concurrency
Batched commits to reduce fsync overhead
Memory cache updated before SQLite to minimize lock time
Specialized batch operations for telemetry

Configuration

Cache behavior is controlled by configuration options:

Option	Default	Description
`cache_dir`	Platform-specific	Cache directory location
`enable_cache`	`True`	Enable/disable caching
`cache_commit_interval`	`25`	Writes before SQLite commit
`sqlite_timeout`	`30.0`	SQLite connection timeout (seconds)
`memory_cache_max_items`	`1024`	Max items in memory cache
`memory_telemetry_cache_max_items`	`2048`	Max telemetry items in memory

See Configuration for details.

Examples

Basic Usage

import tif1

cache = tif1.get_cache()

# Store data
cache.set('my-key', {'data': [1, 2, 3]})

# Retrieve data
data = cache.get('my-key')
print(data)  # {'data': [1, 2, 3]}

Async Operations

import asyncio
import tif1

async def main():
    cache = tif1.get_cache()
    
    # Store asynchronously
    await cache.set_async('my-key', {'value': 42})
    
    # Retrieve asynchronously
    data = await cache.get_async('my-key')
    print(data)

asyncio.run(main())

Telemetry Caching

import tif1

cache = tif1.get_cache()

# Cache telemetry data
telemetry_data = {'Speed': [100, 120, 150], 'RPM': [8000, 9000, 10000]}
cache.set_telemetry(2024, 'Monaco', 'R', 'VER', 1, telemetry_data)

# Retrieve telemetry
data = cache.get_telemetry(2024, 'Monaco', 'R', 'VER', 1)
if data:
    print(f"Speed: {data['Speed']}")

Batch Telemetry Lookup

import tif1

cache = tif1.get_cache()

# Get multiple telemetry entries at once
driver_laps = [
    ('VER', 1), ('VER', 2), ('VER', 3),
    ('HAM', 1), ('HAM', 2)
]

results = cache.get_telemetry_batch(2024, 'Monaco', 'R', driver_laps)

print(f"Found {len(results)} cached entries")
for (driver, lap), data in results.items():
    print(f"  {driver} lap {lap}: {list(data.keys())}")

Check Session Data

import tif1

cache = tif1.get_cache()

if cache.has_session_data(2024, 'Monaco', 'R'):
    print('Monaco race data is cached')
else:
    print('Need to fetch Monaco race data')

Custom Cache Location

from pathlib import Path
import tif1

config = tif1.get_config()
config.set('cache_dir', '~/my-project/.tif1-cache')

cache = tif1.get_cache()
print(f"Cache location: {cache.cache_dir}")

Clear Cache

import tif1

cache = tif1.get_cache()
cache.clear()
print('All cached data cleared')

Source Code

For the complete implementation, see cache.py:42-530 in the source code.

Core API

Classes

Configuration

Data Types

Exceptions

Documentation Index

​Overview

​Functions

​get_cache()

​Cache Class

​Constructor

​Attributes

​Methods

​General Cache Operations

​get()

​get_async()

​set()

​set_async()

​Telemetry-Specific Operations

​get_telemetry()

​get_telemetry_async()

​get_telemetry_batch()

​get_telemetry_batch_async()

​set_telemetry()

​set_telemetry_async()

​Cache Management

​has_session_data()

​clear()

​close()

​Default Cache Directory

​Cache Architecture

​Two-Tier Caching

​Thread Safety

​Performance Optimizations

​Configuration

​Examples

​Basic Usage

​Async Operations

​Telemetry Caching

​Batch Telemetry Lookup

​Check Session Data

​Custom Cache Location

​Clear Cache

​Source Code

Build docs developers (and LLMs) love

Overview

Functions

get_cache()

Cache Class

Constructor

Attributes

Methods

General Cache Operations

get()

get_async()

set()

set_async()

Telemetry-Specific Operations

get_telemetry()

get_telemetry_async()

get_telemetry_batch()

get_telemetry_batch_async()

set_telemetry()

set_telemetry_async()

Cache Management

has_session_data()

clear()

close()

Default Cache Directory

Cache Architecture

Two-Tier Caching

Thread Safety

Performance Optimizations

Configuration

Examples

Basic Usage

Async Operations

Telemetry Caching

Batch Telemetry Lookup

Check Session Data

Custom Cache Location

Clear Cache

Source Code