V0.3.0 release

.github/workflows/release.yml

@@ -1,7 +1,7 @@
 # Publish package on main branch if it's tagged with 'v*'
 # Ref: https://github.community/t/run-workflow-on-push-tag-on-specific-branch/17519
 
-name: build & release
+name: release
 
 # Controls when the action will run.
 on:

.github/workflows/test.yml

@@ -31,7 +31,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install .[test]
+        pip install .[local,test]
     - name: Lint with ruff
       run: |
         # stop the build if there are Python syntax errors or undefined names

HISTORY.md

@@ -1,5 +1,11 @@
 # History
 
+## 0.3.0 (2025-09-13)
+
+* Fix so local caching via TOML works.
+* `get_secret()`: Add `force_refresh` and `raw` parameters
+* Attempt to `JSON.loads` secret string from AWS Secrets Manager by default.
+
 ## 0.2.0 (2025-09-13)
 
 * Add core caching logic.

README.md

@@ -1,6 +1,9 @@
 # Secrets Cache
 
-![PyPI version](https://img.shields.io/pypi/v/secrets-cache.svg)
+[![PyPI version](https://img.shields.io/pypi/v/secrets-cache.svg)](https://pypi.org/project/secrets-cache/)
+[![PyPI license](https://img.shields.io/pypi/l/secrets-cache.svg)](https://pypi.org/project/secrets-cache/)
+[![PyPI Python versions](https://img.shields.io/pypi/pyversions/secrets-cache.svg)](https://pypi.org/project/secrets-cache/)
+[![GitHub Actions](https://github.com/rnag/py-secrets-cache/actions/workflows/release.yml/badge.svg)](https://github.com/rnag/py-secrets-cache/actions/workflows/release.yml)
 [![Documentation Status](https://readthedocs.org/projects/secrets-cache/badge/?version=latest)](https://secrets-cache.readthedocs.io/en/latest/?version=latest)
 
 Cache secrets locally from AWS Secrets Manager and other secret stores, with optional local caching for development or Lambda-friendly usage.
@@ -17,40 +20,48 @@ Install the base package (minimal, Lambda-friendly):
 pip install secrets-cache[lambda]
 ````
 
-Install with **local cache support** (TOML) for testing / development:
+For local development or testing (with local TOML caching, AWS SDK):
 
 ```bash
 pip install secrets-cache[local]
 ```
 
-Install with CLI support:
+Optional CLI tools:
 
 ```bash
 pip install secrets-cache[cli]
 ```
 
-You can also combine extras:
+## Usage
 
-```bash
-pip install "secrets-cache[local,cli]"
+### Fetch a secret from AWS Secrets Manager
+
+```python
+from secrets_cache import get_secret
+
+# Returns JSON-decoded dict if possible
+db_creds = get_secret("prod/AppBeta/MySQL")
+
+# Returns raw string
+raw_value = get_secret("prod/AppBeta/MySQL", raw=True)
+
+# Force refresh from AWS, ignoring cache
+fresh_value = get_secret("prod/AppBeta/MySQL", force_refresh=True)
 ```
 
-## Usage
+### Fetch a parameter from AWS SSM Parameter Store
 
 ```python
-from secrets_cache import get_secret, get_param
-
-# Get a secret from AWS Secrets Manager
-my_secret = get_secret("my-secret-name", region="us-east-1")
+from secrets_cache import get_param
 
-# Get a parameter from AWS SSM Parameter Store
-my_param = get_param("/my/parameter/name", region="us-east-1")
+api_url = get_param("prod/AppBeta/API_URL")
 ```
 
 **Notes:**
 
-* By default, secrets are cached **in memory** to reduce repeated AWS calls.
-* If `local` extra is installed, secrets are also stored in `~/.secrets_cache.toml` for local caching.
+* Secrets and parameters are **cached in-memory** and optionally in a **local TOML file** (`~/.secrets_cache.toml`) for repeated calls.
+* Default cache TTL is **1 week** (configurable via `SECRETS_CACHE_TTL` environment variable).
+* AWS region defaults to `AWS_REGION` environment variable or `us-east-1`.
 * Module-level caches persist across **warm AWS Lambda invocations**, so repeated calls in the same container are very fast.
 
 ## Features

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "secrets-cache"
-version = "0.2.0"
+version = "0.3.0"
 description = "Cache secrets locally from AWS Secrets Manager and other secret stores."
 readme = "README.md"
 authors = [
@@ -38,9 +38,9 @@ cli = [
     'typer',  # CLI / optional but handy
 ]
 local = [
-    'boto3>=1.26',  # AWS SDK for fetching secrets
+    'boto3>=1.26',   # AWS SDK for fetching secrets
+    'tomli-w>=1.0',  # TOML writer for Python <3.11
     'tomli>=2.0; python_version<"3.11"',  # TOML reader for Python <3.11
-    'tomli-w>=1.0; python_version<"3.11"',  # TOML writer for Python <3.11
 ]
 lambda = []              # Lambda-friendly, skips boto3
 test = [

src/secrets_cache/_aws.py

@@ -1,37 +1,17 @@
-"""Main module."""
-
-from pathlib import Path
+"""AWS module."""
+from json import loads
 
 import boto3
+from botocore.exceptions import ClientError
 
 from ._cache import cached_fetch
-
-
-# Optional: import TOML only if available
-try:
-    import tomllib  # Python 3.11+
-except ImportError:
-    try:
-        import tomli as tomllib  # Python 3.10
-    except ImportError:
-        tomllib = None
-
-try:
-    import tomli_w
-except ImportError:
-    tomli_w = None
-
+from ._constants import DEFAULT_CACHE_TTL, DEFAULT_AWS_REGION
 
 # Module-level caches
-_secret_cache = {}
-_param_cache = {}
 _boto_clients = {}
 
-# TOML cache file (optional)
-_CACHE_FILE = Path.home() / '.secrets_cache.toml'
 
-
-def get_boto_client(service: str, region: str = 'us-east-1'):
+def get_boto_client(service: str, region: str):
     """Cache boto3 clients per service/region."""
     key = service, region
     _client = _boto_clients.get(key)
@@ -40,49 +20,65 @@ def get_boto_client(service: str, region: str = 'us-east-1'):
     return _client
 
 
-def _read_toml_cache():
-    if not tomllib or not _CACHE_FILE.exists():
-        return {}
-    with _CACHE_FILE.open('rb') as f:
-        return tomllib.load(f)
-
-
-def _write_toml_cache(data):
-    if not tomli_w or not _CACHE_FILE.parent.exists():
-        return
-    with _CACHE_FILE.open('wb') as f:
-        tomli_w.dump(data, f)
+def get_secret(name: str,
+               region: str = DEFAULT_AWS_REGION,
+               ttl: int = DEFAULT_CACHE_TTL,
+               force_refresh: bool = False,
+               raw: bool = False) -> str | bytes | dict:
+    """Get secret from AWS Secrets Manager with optional caching."""
+    value = cached_fetch('secretsmanager', name, region, _fetch_secret, ttl, force_refresh)
 
+    if raw:
+        return value
 
-def get_secret(name: str, region: str = 'us-east-1', ttl: int = 7 * 24 * 3600):
-    """Get secret from AWS Secrets Manager with optional caching."""
-    return cached_fetch(_secret_cache, name, region, _fetch_secret, ttl)
+    # Try to parse JSON (most Secrets Manager use case)
+    try:
+        return loads(value)
+    except (ValueError, TypeError):
+        return value
 
 
-def get_param(name: str, region: str = 'us-east-1', ttl: int = 7 * 24 * 3600):
+def get_param(name: str,
+              region: str = DEFAULT_AWS_REGION,
+              ttl: int = DEFAULT_CACHE_TTL,
+              force_refresh: bool = False) -> str:
     """Get parameter from AWS SSM Parameter Store with optional caching."""
-    return cached_fetch(_param_cache, name, region, _fetch_param, ttl)
+    return cached_fetch('ssm', name, region, _fetch_param, ttl, force_refresh)
 
 
-def _fetch_secret(name: str, region: str = 'us-east-1'):
+def _fetch_secret(name: str, region: str) -> str | bytes | None:
     client = get_boto_client('secretsmanager', region)
-    resp = client.get_secret_value(SecretId=name)
-    value = resp['SecretString']
 
-    # Update local TOML cache if available
-    data = _read_toml_cache()
-    data[name] = value
-    _write_toml_cache(data)
-    return value
-
-
-def _fetch_param(name: str, region: str = 'us-east-1'):
+    try:
+        resp = client.get_secret_value(
+            SecretId=name
+        )
+    except ClientError as e:
+        # For a list of exceptions thrown, see
+        # https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html
+        err_code = e.response['Error']['Code']
+        if err_code == 'ResourceNotFoundException':
+            print(f'The requested secret {name} was not found')
+        elif err_code == 'InvalidRequestException':
+            print('The request was invalid due to:', e)
+        elif err_code == 'InvalidParameterException':
+            print('The request had invalid params:', e)
+        elif err_code == 'DecryptionFailure':
+            print('The requested secret can\'t be decrypted using the provided KMS key:', e)
+        elif err_code == 'InternalServiceError':
+            print('An error occurred on service side:', e)
+    else:
+        # Secrets Manager decrypts the secret value using the associated KMS CMK
+        # Depending on whether the secret was a string or binary, only one of these fields will be populated
+        if (secret := resp.get('SecretString')) is not None:
+            return secret
+        return resp['SecretBinary']
+
+
+def _fetch_param(name: str, region: str):
     client = get_boto_client('ssm', region)
+
     resp = client.get_parameter(Name=name, WithDecryption=True)
-    value = resp['Parameter']['Value']
 
-    # Update local TOML cache if available
-    data = _read_toml_cache()
-    data[name] = value
-    _write_toml_cache(data)
-    return value
+    param = resp['Parameter']['Value']
+    return param

src/secrets_cache/_cache.py

@@ -1,13 +1,90 @@
+"""Caching module."""
+from pathlib import Path
 from time import time
 
 
-def cached_fetch(cache: dict, key: str, region: str, fetcher, ttl: int):
-    """Fetch from cache or call fetcher."""
-    now = time()
-    entry = cache.get(key)
+# TOML cache file (optional)
+_CACHE_FILE = Path.home() / '.secrets_cache.toml'
+
+
+# Lazy-loaded cache
+_cache = None
+
+
+# Optional: import TOML only if available
+try:
+    import tomllib  # Python 3.11+
+except ImportError:
+    try:
+        # noinspection SpellCheckingInspection
+        import tomli as tomllib  # Python 3.10
+    except ImportError:
+        # noinspection SpellCheckingInspection
+        tomllib = None
+
+try:
+    import tomli_w
+except ImportError:
+    tomli_w = None
+
+
+def _load_cache() -> dict:
+    global _cache
+
+    if _cache is None:
+        if tomllib and _CACHE_FILE.exists():
+            with _CACHE_FILE.open('rb') as f:
+                _cache = tomllib.load(f)
+        else:
+            _cache = {}
+
+    return _cache
+
+
+def _save_cache() -> None:
+    if (_cache is None
+            or not tomli_w
+            or not _CACHE_FILE.parent.exists()):
+        return
+
+    with _CACHE_FILE.open('wb') as f:
+        tomli_w.dump(_cache, f)  # type: ignore
+
+
+def get_cached_value(service: str, name: str, now: float, ttl: int):
+    data = _load_cache()
+    service_cache = data.get(service, {})
+    entry = service_cache.get(name)
+
     if entry and (now - entry['fetched_at'] < ttl):
         return entry['value']
 
+    return None
+
+
+def update_cache(service: str, name: str, value: str, now: float):
+    data = _load_cache()
+    final_value = {'value': value, 'fetched_at': now}
+
+    service_cache = data.get(service)
+
+    if service_cache is None:
+        data[service] = {name: final_value}
+    else:
+        service_cache[name] = final_value
+
+    _save_cache()
+
+
+def cached_fetch(service: str, key: str, region: str, fetcher, ttl: int, force_refresh: bool):
+    """Fetch from cache or call fetcher."""
+    now = int(time())
+
+    if not force_refresh:
+        if (value := get_cached_value(service, key, now, ttl)) is not None:
+            return value
+
     value = fetcher(key, region)
-    cache[key] = {'value': value, 'fetched_at': now}
+    if value is not None:
+        update_cache(service, key, value, now)
     return value