lifecycle

beautyspot.lifecycle モジュールは、データの保持期間（Retention）の解析、および関数名に基づいたポリシーの解決を担います。 現在は 完全修飾名（module.qualname）を優先 して解決し、マッチしない場合のみ短い関数名にフォールバックします。

beautyspot.lifecycle

LifecyclePolicy

Manages data retention policies based on function names.

Source code in src/beautyspot/lifecycle.py

class LifecyclePolicy:
    """
    Manages data retention policies based on function names.
    """

    def __init__(
        self,
        rules: List[Rule],
        default_retention: Union[str, timedelta, int, float, None] = "30d",
    ):
        if isinstance(default_retention, _ForeverSentinel):
            raise ValidationError(
                "Retention.FOREVER cannot be used as default_retention. "
                "Use it on individual @spot.mark(retention=Retention.FOREVER) instead."
            )
        self.rules = rules
        self._default_retention = parse_retention(default_retention)

    def resolve(self, func_name: str) -> Optional[timedelta]:
        """
        Find the first matching rule for the given function name.
        Returns the retention timedelta, or the default retention if no match.
        """
        for rule in self.rules:
            if fnmatch.fnmatch(func_name, rule.pattern):
                return parse_retention(rule.retention)
        return self._default_retention

    def resolve_with_fallback(
        self, func_identifier: str, func_name: str
    ) -> Optional[timedelta]:
        """
        Resolve retention using the fully-qualified identifier first, then
        fall back to the short function name for backward compatibility.
        """
        for rule in self.rules:
            if fnmatch.fnmatch(func_identifier, rule.pattern):
                return parse_retention(rule.retention)

        for rule in self.rules:
            if fnmatch.fnmatch(func_name, rule.pattern):
                return parse_retention(rule.retention)

        return self._default_retention

    @classmethod
    def default(cls) -> "LifecyclePolicy":
        """Default policy: 30-day retention."""
        return cls(rules=[], default_retention="30d")

default() classmethod

Default policy: 30-day retention.

Source code in src/beautyspot/lifecycle.py

@classmethod
def default(cls) -> "LifecyclePolicy":
    """Default policy: 30-day retention."""
    return cls(rules=[], default_retention="30d")

resolve(func_name)

Find the first matching rule for the given function name. Returns the retention timedelta, or the default retention if no match.

Source code in src/beautyspot/lifecycle.py

def resolve(self, func_name: str) -> Optional[timedelta]:
    """
    Find the first matching rule for the given function name.
    Returns the retention timedelta, or the default retention if no match.
    """
    for rule in self.rules:
        if fnmatch.fnmatch(func_name, rule.pattern):
            return parse_retention(rule.retention)
    return self._default_retention

resolve_with_fallback(func_identifier, func_name)

Resolve retention using the fully-qualified identifier first, then fall back to the short function name for backward compatibility.

Source code in src/beautyspot/lifecycle.py

def resolve_with_fallback(
    self, func_identifier: str, func_name: str
) -> Optional[timedelta]:
    """
    Resolve retention using the fully-qualified identifier first, then
    fall back to the short function name for backward compatibility.
    """
    for rule in self.rules:
        if fnmatch.fnmatch(func_identifier, rule.pattern):
            return parse_retention(rule.retention)

    for rule in self.rules:
        if fnmatch.fnmatch(func_name, rule.pattern):
            return parse_retention(rule.retention)

    return self._default_retention

Retention

Constants for retention policies.

Attributes:

Name	Type	Description
INDEFINITE		ライフサイクルポリシーに委ねるデフォルト値 (None)。 ポリシーが設定されている場合はそのルールに従い、 未設定の場合は無期限保持となります。
FOREVER	_ForeverSentinel	ライフサイクルポリシーを明示的にバイパスし、 このキャッシュエントリを常に無期限保持することを宣言します。 @spot.mark(retention=Retention.FOREVER) で使用します。

Source code in src/beautyspot/lifecycle.py

class Retention:
    """Constants for retention policies.

    Attributes:
        INDEFINITE: ライフサイクルポリシーに委ねるデフォルト値 (None)。
            ポリシーが設定されている場合はそのルールに従い、
            未設定の場合は無期限保持となります。
        FOREVER: ライフサイクルポリシーを明示的にバイパスし、
            このキャッシュエントリを常に無期限保持することを宣言します。
            ``@spot.mark(retention=Retention.FOREVER)`` で使用します。
    """

    INDEFINITE = None
    FOREVER: _ForeverSentinel = _FOREVER

Rule dataclass

A rule defining retention policy based on function name pattern.

Source code in src/beautyspot/lifecycle.py

@dataclass
class Rule:
    """
    A rule defining retention policy based on function name pattern.
    """

    pattern: str
    retention: Union[str, timedelta, int, None]

parse_retention(value)

Helper function to normalize retention specification to timedelta. None means 'indefinite' (defers to lifecycle policy). int is treated as 'seconds'.

Note

_ForeverSentinel (Retention.FOREVER) は本関数では処理しません。 呼び出し元 (Spot._calculate_expires_at) で事前にチェックしてください。

Source code in src/beautyspot/lifecycle.py

def parse_retention(
    value: RetentionSpec
) -> Optional[timedelta]:
    """
    Helper function to normalize retention specification to timedelta.
    None means 'indefinite' (defers to lifecycle policy).
    int is treated as 'seconds'.

    Note:
        ``_ForeverSentinel`` (``Retention.FOREVER``) は本関数では処理しません。
        呼び出し元 (``Spot._calculate_expires_at``) で事前にチェックしてください。
    """
    if value is None:
        return None

    if isinstance(value, timedelta):
        if value.total_seconds() <= 0:
            raise ValidationError(f"Retention timedelta must be positive, got {value}.")
        return value

    if isinstance(value, (int, float)):
        if value <= 0:
            raise ValidationError(
                f"Retention must be a positive number (seconds), got {value}."
            )
        return timedelta(seconds=value)

    if isinstance(value, str):
        match = _TIME_PATTERN.match(value)
        if not match:
            raise ValidationError(
                f"Invalid retention format: '{value}'. Use format like '7d', '12h', '30m', '10s'."
            )

        amount, unit = int(match.group(1)), match.group(2)
        if amount <= 0:
            raise ValidationError(
                f"Retention duration must be positive, got '{value}'."
            )
        if unit == "d":
            return timedelta(days=amount)
        elif unit == "h":
            return timedelta(hours=amount)
        elif unit == "m":
            return timedelta(minutes=amount)
        elif unit == "s":
            return timedelta(seconds=amount)

    raise ValidationError(
        f"Retention must be str, int, float, or timedelta, got {type(value)}"
    )

主要な構成要素

1. Retention 解決ロジック (parse_retention)

ユーザーが指定した多様な形式（文字列、整数、timedelta）を、内部で扱う timedelta オブジェクトへと正規化します。

• サポートされる形式:
• 文字列: "7d" (日), "12h" (時間), "30m" (分) の形式。
• 整数: 秒単位として扱われます。
• timedelta: そのまま利用されます。
• None: Retention.INDEFINITE (無期限) として扱われます。

2. Rule クラス

特定の関数名パターンに対して適用する保持期間を定義するデータクラスです。

属性	型	説明
pattern	str	fnmatch 形式のワイルドカードパターン（例: api_*）
retention	Union[str, int, timedelta, None]	適用する保持期間

3. LifecyclePolicy クラス

ルールの集合を管理し、実行時に特定の関数名に対してどのルールを適用するかを決定（Resolve）します。

• 解決アルゴリズム: rules リストの 先頭から順に パターンマッチングを行い、最初にマッチしたルールの保持期間を返します。
• 適用順序: 完全修飾名（module.qualname）→ 短い関数名（func_name）の順でマッチングします。
• デフォルト挙動: マッチするルールがない場合、またはルールリストが空の場合は Retention.INDEFINITE（無期限）が返されます。

型定義と定数

Retention クラス

• INDEFINITE: None（無期限保持を明示するためのエイリアス）。

注意事項

• パターンマッチング: 内部で fnmatch.fnmatch を使用しているため、大文字小文字の区別は OS のファイルシステムルールに依存する可能性があります。一貫性を保つため、関数名の命名規則をプロジェクト内で統一することを推奨します。
• バリデーション: parse_retention は、不正なフォーマット（例: "1year" や "10s" ※現状秒単位の文字指定は未サポート）が渡された場合、beautyspot.exceptions.ValidationError を送出します。