1468af1391
- Created FastAPI application with SQLite database integration - Implemented OpenWeatherMap client with caching - Added endpoints for current weather, forecasts, and history - Included comprehensive error handling and validation - Set up Alembic migrations - Created detailed README with usage examples generated with BackendIM... (backend.im)
99 lines
2.9 KiB
Python
99 lines
2.9 KiB
Python
from typing import Any, Dict, Optional, Callable, TypeVar, Awaitable
|
|
from datetime import datetime, timedelta
|
|
import hashlib
|
|
import json
|
|
|
|
# In-memory cache
|
|
cache: Dict[str, Dict[str, Any]] = {}
|
|
|
|
T = TypeVar("T")
|
|
|
|
class Cache:
|
|
@staticmethod
|
|
def _generate_key(prefix: str, *args: Any, **kwargs: Any) -> str:
|
|
"""
|
|
Generate a unique cache key based on the function arguments.
|
|
"""
|
|
key_parts = [prefix]
|
|
|
|
# Add positional args to key
|
|
for arg in args:
|
|
key_parts.append(str(arg))
|
|
|
|
# Add keyword args to key (sorted by key to ensure consistency)
|
|
for k, v in sorted(kwargs.items()):
|
|
key_parts.append(f"{k}={v}")
|
|
|
|
# Join all parts and hash
|
|
key_data = ":".join(key_parts)
|
|
return hashlib.md5(key_data.encode()).hexdigest()
|
|
|
|
@staticmethod
|
|
def set(key: str, value: Any, expire_seconds: int = 1800) -> None:
|
|
"""
|
|
Store a value in the cache with an expiration time.
|
|
"""
|
|
expiry = datetime.utcnow() + timedelta(seconds=expire_seconds)
|
|
cache[key] = {
|
|
"value": value,
|
|
"expiry": expiry
|
|
}
|
|
|
|
@staticmethod
|
|
def get(key: str) -> Optional[Any]:
|
|
"""
|
|
Retrieve a value from the cache if it exists and hasn't expired.
|
|
"""
|
|
if key not in cache:
|
|
return None
|
|
|
|
cache_item = cache[key]
|
|
if cache_item["expiry"] < datetime.utcnow():
|
|
# Remove expired item
|
|
del cache[key]
|
|
return None
|
|
|
|
return cache_item["value"]
|
|
|
|
@staticmethod
|
|
def invalidate(key: str) -> None:
|
|
"""
|
|
Remove a specific item from the cache.
|
|
"""
|
|
if key in cache:
|
|
del cache[key]
|
|
|
|
@staticmethod
|
|
def clear_all() -> None:
|
|
"""
|
|
Clear the entire cache.
|
|
"""
|
|
cache.clear()
|
|
|
|
def cached(prefix: str, expire_seconds: int = 1800):
|
|
"""
|
|
Decorator for async functions to cache their results.
|
|
|
|
Args:
|
|
prefix: Prefix for the cache key to avoid collisions
|
|
expire_seconds: Time in seconds until the cached result expires
|
|
"""
|
|
def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
|
|
async def wrapper(*args: Any, **kwargs: Any) -> T:
|
|
# Generate a unique key based on function arguments
|
|
key = Cache._generate_key(prefix, *args, **kwargs)
|
|
|
|
# Check if we have a cached result
|
|
cached_result = Cache.get(key)
|
|
if cached_result is not None:
|
|
return cached_result
|
|
|
|
# Call the original function
|
|
result = await func(*args, **kwargs)
|
|
|
|
# Cache the result
|
|
Cache.set(key, result, expire_seconds)
|
|
|
|
return result
|
|
return wrapper
|
|
return decorator |